diff --git a/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts b/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts index a4f47e25ad3..7c0cab056e0 100644 --- a/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts +++ b/packages/@n8n/engine/src/database/__tests__/workflow-execution.integration.test.ts @@ -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); diff --git a/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts b/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts index 06fda3549d8..bcd86cabb67 100644 --- a/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts +++ b/packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts @@ -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({ diff --git a/packages/@n8n/engine/src/database/typeorm-execution-view-store.ts b/packages/@n8n/engine/src/database/typeorm-execution-view-store.ts index 9913853b564..6fc23adcebc 100644 --- a/packages/@n8n/engine/src/database/typeorm-execution-view-store.ts +++ b/packages/@n8n/engine/src/database/typeorm-execution-view-store.ts @@ -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 { - 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 { + 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 { + 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 { - 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, + }; +} diff --git a/packages/@n8n/engine/src/execution/__tests__/execution-query.service.test.ts b/packages/@n8n/engine/src/execution/__tests__/execution-query.service.test.ts index 4491be887a0..b335824ea58 100644 --- a/packages/@n8n/engine/src/execution/__tests__/execution-query.service.test.ts +++ b/packages/@n8n/engine/src/execution/__tests__/execution-query.service.test.ts @@ -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 { 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, + ); }); }); diff --git a/packages/@n8n/engine/src/execution/execution-query.service.ts b/packages/@n8n/engine/src/execution/execution-query.service.ts index f32059ab7d1..de3d988f90b 100644 --- a/packages/@n8n/engine/src/execution/execution-query.service.ts +++ b/packages/@n8n/engine/src/execution/execution-query.service.ts @@ -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 { - 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 { + return await this.viewStore.loadExecutionWithStepsView(id); } } diff --git a/packages/@n8n/engine/src/execution/execution-view-store.ts b/packages/@n8n/engine/src/execution/execution-view-store.ts index d948b3ef336..874948277e4 100644 --- a/packages/@n8n/engine/src/execution/execution-view-store.ts +++ b/packages/@n8n/engine/src/execution/execution-view-store.ts @@ -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; - /** Read views of every step of the execution, oldest first. Empty if it has none yet. */ - loadStepViews(executionId: string): Promise; + /** + * 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; } diff --git a/packages/@n8n/engine/src/execution/index.ts b/packages/@n8n/engine/src/execution/index.ts index 44579664458..0f912fa83d5 100644 --- a/packages/@n8n/engine/src/execution/index.ts +++ b/packages/@n8n/engine/src/execution/index.ts @@ -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'; diff --git a/packages/@n8n/engine/src/index.ts b/packages/@n8n/engine/src/index.ts index f4c21dd22e9..a2d00e067ce 100644 --- a/packages/@n8n/engine/src/index.ts +++ b/packages/@n8n/engine/src/index.ts @@ -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 { diff --git a/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts b/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts index ba7adc4a152..f92a595d926 100644 --- a/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts +++ b/packages/@n8n/engine/src/server/__tests__/workflow-executions.integration.test.ts @@ -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); diff --git a/packages/@n8n/engine/src/server/api.types.ts b/packages/@n8n/engine/src/server/api.types.ts index dad001a6a3e..2b938779fb0 100644 --- a/packages/@n8n/engine/src/server/api.types.ts +++ b/packages/@n8n/engine/src/server/api.types.ts @@ -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[]; -} diff --git a/packages/@n8n/engine/src/server/index.ts b/packages/@n8n/engine/src/server/index.ts index 4ac770a053d..a4e339de031 100644 --- a/packages/@n8n/engine/src/server/index.ts +++ b/packages/@n8n/engine/src/server/index.ts @@ -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'; diff --git a/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts b/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts index d03ec527348..7faa3eb3f2c 100644 --- a/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts +++ b/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts @@ -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)); }; } diff --git a/packages/@n8n/engine/src/server/routes/workflow-executions.ts b/packages/@n8n/engine/src/server/routes/workflow-executions.ts index aa8f562bafb..f210ed011aa 100644 --- a/packages/@n8n/engine/src/server/routes/workflow-executions.ts +++ b/packages/@n8n/engine/src/server/routes/workflow-executions.ts @@ -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; } diff --git a/packages/@n8n/node-engine-compatibility/src/__tests__/v1-execution-read.test.ts b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-execution-read.test.ts new file mode 100644 index 00000000000..7b2c946edfe --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/__tests__/v1-execution-read.test.ts @@ -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 => ({ + 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(['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 }]); + } + }); + }); +}); diff --git a/packages/@n8n/node-engine-compatibility/src/index.ts b/packages/@n8n/node-engine-compatibility/src/index.ts index fe2769c919b..0fb5d593657 100644 --- a/packages/@n8n/node-engine-compatibility/src/index.ts +++ b/packages/@n8n/node-engine-compatibility/src/index.ts @@ -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, diff --git a/packages/@n8n/node-engine-compatibility/src/v1-adapters.ts b/packages/@n8n/node-engine-compatibility/src/v1-adapters.ts index 7f0d621e224..13ce3e0a00a 100644 --- a/packages/@n8n/node-engine-compatibility/src/v1-adapters.ts +++ b/packages/@n8n/node-engine-compatibility/src/v1-adapters.ts @@ -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> { - const namesById = new Map(graph.nodes.map((node) => [node.id, node.name])); + const namesById = nodeNamesById(graph); const sourcesByNodeId = new Map>(); - 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; } diff --git a/packages/@n8n/node-engine-compatibility/src/v1-execution-read.ts b/packages/@n8n/node-engine-compatibility/src/v1-execution-read.ts new file mode 100644 index 00000000000..75ae583eabd --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/v1-execution-read.ts @@ -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([ + ['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; + /** Forward edges only, by target node. */ + edgesByTarget: Map; + loops: WorkflowLoop[]; + lastIterationByNodeId: Map; +} + +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(); + 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 { + 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, +): 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; +} diff --git a/packages/@n8n/node-engine-compatibility/src/v1-run-data.ts b/packages/@n8n/node-engine-compatibility/src/v1-run-data.ts new file mode 100644 index 00000000000..8caaa181da2 --- /dev/null +++ b/packages/@n8n/node-engine-compatibility/src/v1-run-data.ts @@ -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 { + 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 { + const edgesByTarget = new Map(); + + 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, + previousNodeRun?: (edge: GraphEdge) => number, +): Array { + const sources: Array = []; + + 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 = [], +): ITaskData { + return { + startTime: 0, + executionTime: 0, + executionIndex, + source, + data: { [MAIN_CONNECTION_TYPE]: [[]] }, + }; +} diff --git a/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts b/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts index c94d4f6324b..7f2a9f13d65 100644 --- a/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts +++ b/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts @@ -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 => ({ + 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 => ({ 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({ 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` 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); diff --git a/packages/cli/src/executions/engine-v2-execution-reader.service.ts b/packages/cli/src/executions/engine-v2-execution-reader.service.ts index f6ab8de7686..bc581eb1e54 100644 --- a/packages/cli/src/executions/engine-v2-execution-reader.service.ts +++ b/packages/cli/src/executions/engine-v2-execution-reader.service.ts @@ -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 { // 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: [] }, diff --git a/packages/cli/src/executions/execution-data/types.ts b/packages/cli/src/executions/execution-data/types.ts index 26bd18a0507..0efeed7c608 100644 --- a/packages/cli/src/executions/execution-data/types.ts +++ b/packages/cli/src/executions/execution-data/types.ts @@ -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; }; diff --git a/packages/cli/src/executions/execution-persistence.ts b/packages/cli/src/executions/execution-persistence.ts index 727df9d6b15..12ee2cf6fa5 100644 --- a/packages/cli/src/executions/execution-persistence.ts +++ b/packages/cli/src/executions/execution-persistence.ts @@ -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 { 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, - ): WorkflowSnapshot { - const { id, name, nodes, connections, settings, nodeGroups } = workflowData; - return { id, name, nodes, connections, settings, nodeGroups }; - } - private async assembleExecution( entity: ExecutionEntity, bundle: ExecutionDataPayload, diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-data-plane-client.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-data-plane-client.test.ts index d1980009096..3a4a4ea4c6e 100644 --- a/packages/cli/src/modules/engine-v2/__tests__/engine-data-plane-client.test.ts +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-data-plane-client.test.ts @@ -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' }); diff --git a/packages/cli/src/modules/engine-v2/engine-data-plane-client.ts b/packages/cli/src/modules/engine-v2/engine-data-plane-client.ts index cf67a1ee00c..d9cb8f50f2e 100644 --- a/packages/cli/src/modules/engine-v2/engine-data-plane-client.ts +++ b/packages/cli/src/modules/engine-v2/engine-data-plane-client.ts @@ -83,10 +83,15 @@ export class EngineDataPlaneClient implements EngineDataPlaneProvider { return response.body as StartExecutionResult; } - async getExecution(id: ExecutionIdV2): Promise { + async getExecution( + id: ExecutionIdV2, + options?: { includeSteps?: boolean }, + ): Promise { const response = await this.http.request({ 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, diff --git a/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts b/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts index 220f7fbc7fe..bf119545b15 100644 --- a/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts +++ b/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts @@ -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(); + proxy.registerProvider(provider); + + await proxy.getExecution(executionId, { includeSteps: true }); + + expect(provider.getExecution).toHaveBeenCalledWith(executionId, { includeSteps: true }); }); }); diff --git a/packages/cli/src/services/engine-data-plane-proxy.service.ts b/packages/cli/src/services/engine-data-plane-proxy.service.ts index 95e7b892a6a..2ed64a50f72 100644 --- a/packages/cli/src/services/engine-data-plane-proxy.service.ts +++ b/packages/cli/src/services/engine-data-plane-proxy.service.ts @@ -13,8 +13,15 @@ import type { ExecutionIdV2 } from '@/executions/execution-id'; export interface EngineDataPlaneProvider { startExecution(request: StartExecutionRequest): Promise; - /** `undefined` when the data plane holds no execution under that id. */ - getExecution(id: ExecutionIdV2): Promise; + /** + * `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; } /** @@ -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 { + async getExecution( + id: ExecutionIdV2, + options?: { includeSteps?: boolean }, + ): Promise { if (!this.provider) return undefined; - return await this.provider.getExecution(id); + return await this.provider.getExecution(id, options); } } diff --git a/packages/cli/test/integration/executions.controller.test.ts b/packages/cli/test/integration/executions.controller.test.ts index 546d2f178b7..7d801b46439 100644 --- a/packages/cli/test/integration/executions.controller.test.ts +++ b/packages/cli/test/integration/executions.controller.test.ts @@ -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); });