From dac4f63f1e940bfadd3441e9622e10c6ac914d20 Mon Sep 17 00:00:00 2001 From: Tomi Turtiainen <10324676+tomi@users.noreply.github.com> Date: Wed, 16 Sep 2026 09:06:07 +0000 Subject: [PATCH] feat(engine): Add endpoints to list and count executions in the data plane (#38109) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Stephen Wright Co-authored-by: Svetoslav Dekov Co-authored-by: Claude Fable 5.1 Co-authored-by: n8n-cat-bot[bot] <283985454+n8n-cat-bot[bot]@users.noreply.github.com> Co-authored-by: n8n-cat-bot[bot] Co-authored-by: Andreas Fitzek Co-authored-by: Elias Meire Co-authored-by: Mutasem Aldmour <4711238+mutdmour@users.noreply.github.com> Co-authored-by: Riqwan Thamir Co-authored-by: Cursor Co-authored-by: Guillaume Jacquart Co-authored-by: n8n-assistant[bot] <100856346+n8n-assistant[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Co-authored-by: Matsu Co-authored-by: Raúl Gómez Morales Co-authored-by: subaru <79771445+subaru-hello@users.noreply.github.com> Co-authored-by: Declan Carroll Co-authored-by: Charlie Kolb Co-authored-by: Roman Makowiak Co-authored-by: Tuukka Kantola --- .../database/typeorm-execution-view-store.ts | 58 ++++++++ .../__tests__/execution-query.service.test.ts | 31 ++++ .../src/execution/execution-query.service.ts | 21 +++ .../src/execution/execution-view-store.ts | 31 ++++ packages/@n8n/engine/src/index.ts | 5 + .../workflow-executions.integration.test.ts | 132 ++++++++++++++++++ packages/@n8n/engine/src/server/api.types.ts | 16 +++ packages/@n8n/engine/src/server/index.ts | 5 + .../routes/workflow-executions.handlers.ts | 46 ++++++ .../src/server/routes/workflow-executions.ts | 6 +- 10 files changed, 350 insertions(+), 1 deletion(-) 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 cfcc35b8df3..c8c1b96e2ea 100644 --- a/packages/@n8n/engine/src/database/typeorm-execution-view-store.ts +++ b/packages/@n8n/engine/src/database/typeorm-execution-view-store.ts @@ -1,17 +1,33 @@ import type { Repository, SelectQueryBuilder } from '@n8n/typeorm'; import type { WorkflowExecution, WorkflowStepExecution } from './entities'; +import { UnexpectedError } from '../common'; import { ExecutionNotFoundError } from '../execution/execution-store'; import type { ExecutionViewStore, ExecutionView, ExecutionWithStepsView, StepView, + ExecutionListQuery, + ExecutionListItemView, } from '../execution/execution-view-store'; /** The execution row, with its steps aggregated into one column. */ type ExecutionWithStepsRow = ExecutionView & { steps: StepView[] }; +/** + * The cursor compares `(created_at, id)`, so it only walks that order. A + * status-bucket sort puts newer rows behind the top bucket, and the next page + * drops every one of them: they sort after the cursor row but were never + * reported. Callers pick one or the other. The API rejects the pair with a 400; + * this holds the rule for every other caller. + */ +function assertPageableOrder(query: ExecutionListQuery): void { + if (query.before && query.order?.top) { + throw new UnexpectedError('An execution cursor cannot page a status-first sort'); + } +} + /** * TypeORM-backed `ExecutionViewStore` adapter. It spans both tables, since a * read of an execution and a read of its steps are one concern. @@ -26,6 +42,48 @@ export class TypeOrmExecutionViewStore implements ExecutionViewStore { private readonly steps: Repository, ) {} + async listExecutionViews(query: ExecutionListQuery): Promise { + assertPageableOrder(query); + + const qb = this.buildListQuery(query) + .select('execution.id', 'id') + .addSelect('execution.workflow_id', 'workflowId') + .addSelect('execution.status', 'status') + .addSelect('execution.mode', 'mode') + .addSelect('execution.created_at', 'createdAt') + .addSelect('execution.updated_at', 'updatedAt') + .addSelect('execution.finished_at', 'finishedAt'); + if (query.before) { + qb.andWhere('(execution.created_at, execution.id) < (:createdAt, :id)', query.before); + } + if (query.order?.top) { + qb.orderBy(`(CASE WHEN execution.status = '${query.order.top}' THEN 0 ELSE 1 END)`); + } + return await qb + .addOrderBy('execution.created_at', 'DESC') + .addOrderBy('execution.id', 'DESC') + .limit(query.limit) + .getRawMany(); + } + + async countExecutionViews(query: ExecutionListQuery): Promise { + return await this.buildListQuery(query).getCount(); + } + + private buildListQuery(query: ExecutionListQuery): SelectQueryBuilder { + const qb = this.executions.createQueryBuilder('execution'); + if (query.workflowIds !== 'all') { + qb.andWhere('execution.workflow_id = ANY(:workflowIds)', { workflowIds: query.workflowIds }); + } + if (query.status) qb.andWhere('execution.status = ANY(:statuses)', { statuses: query.status }); + if (query.mode) qb.andWhere('execution.mode = :mode', { mode: query.mode }); + if (query.createdAfter) + qb.andWhere('execution.created_at >= :createdAfter', { createdAfter: query.createdAfter }); + if (query.createdBefore) + qb.andWhere('execution.created_at <= :createdBefore', { createdBefore: query.createdBefore }); + return qb; + } + async loadExecutionView(id: string): Promise { const row: ExecutionView | undefined = await this.selectExecution(id).getRawOne(); if (!row) throw new ExecutionNotFoundError(id); 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 b335824ea58..3d95d5f5848 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 @@ -4,6 +4,7 @@ import { ExecutionQueryService } from '../execution-query.service'; import { ExecutionNotFoundError } from '../execution-store'; import type { ExecutionViewStore, + ExecutionListItemView, ExecutionView, ExecutionWithStepsView, StepView, @@ -11,6 +12,8 @@ import type { function makeViewStore(overrides: Partial = {}): ExecutionViewStore { return { + listExecutionViews: vi.fn().mockResolvedValue([]), + countExecutionViews: vi.fn().mockResolvedValue(0), loadExecutionView: vi.fn(), loadExecutionWithStepsView: vi.fn(), ...overrides, @@ -18,6 +21,34 @@ function makeViewStore(overrides: Partial = {}): ExecutionVi } describe('ExecutionQueryService', () => { + it('uses lookahead to report the next page cursor without including it in the page', async () => { + const rows = [ + { id: '1', createdAt: '2026-01-01T00:00:00.000Z' }, + { id: '2', createdAt: '2026-01-02T00:00:00.000Z' }, + { id: '3', createdAt: '2026-01-03T00:00:00.000Z' }, + ] as unknown as ExecutionListItemView[]; + const store = makeViewStore({ + listExecutionViews: vi.fn().mockResolvedValue(rows), + countExecutionViews: vi.fn().mockResolvedValue(8), + }); + const service = new ExecutionQueryService(store); + await expect( + service.searchExecutions({ workflowIds: ['wf'], limit: 2, includeTotal: true }), + ).resolves.toEqual({ + items: rows.slice(0, 2), + nextCursor: { id: '2', createdAt: '2026-01-02T00:00:00.000Z' }, + total: 8, + }); + expect(store.listExecutionViews).toHaveBeenCalledWith(expect.objectContaining({ limit: 3 })); + }); + + it('does not count when the request does not ask for a total', async () => { + const store = makeViewStore(); + await expect( + new ExecutionQueryService(store).searchExecutions({ workflowIds: 'all', limit: 20 }), + ).resolves.toEqual({ items: [], nextCursor: null }); + expect(store.countExecutionViews).not.toHaveBeenCalled(); + }); it('getExecution loads and returns the execution view', async () => { const execution = { id: 'exec-1' } as ExecutionView; const viewStore = makeViewStore({ diff --git a/packages/@n8n/engine/src/execution/execution-query.service.ts b/packages/@n8n/engine/src/execution/execution-query.service.ts index de3d988f90b..a7bfdffd8d8 100644 --- a/packages/@n8n/engine/src/execution/execution-query.service.ts +++ b/packages/@n8n/engine/src/execution/execution-query.service.ts @@ -2,6 +2,7 @@ import type { ExecutionViewStore, ExecutionView, ExecutionWithStepsView, + ExecutionListQuery, } from './execution-view-store'; /** @@ -12,6 +13,26 @@ import type { export class ExecutionQueryService { constructor(private readonly viewStore: ExecutionViewStore) {} + /** + * Lists executions matching `query`, one page at a time. Fetches one extra + * row over the requested limit, so the last in-page row's own cursor can be + * reported as `nextCursor` without a second query. + */ + async searchExecutions(query: ExecutionListQuery) { + const { limit } = query; + const [rows, total] = await Promise.all([ + this.viewStore.listExecutionViews({ ...query, limit: limit + 1 }), + query.includeTotal ? this.viewStore.countExecutionViews(query) : undefined, + ]); + const items = rows.slice(0, limit); + const last = items.at(-1); + return { + items, + nextCursor: rows.length > limit && last ? { createdAt: last.createdAt, id: last.id } : null, + ...(total !== undefined ? { total } : {}), + }; + } + /** @throws {ExecutionNotFoundError} if absent. */ async getExecution(id: string): Promise { return await this.viewStore.loadExecutionView(id); diff --git a/packages/@n8n/engine/src/execution/execution-view-store.ts b/packages/@n8n/engine/src/execution/execution-view-store.ts index a2356643930..0c6dfa9bb36 100644 --- a/packages/@n8n/engine/src/execution/execution-view-store.ts +++ b/packages/@n8n/engine/src/execution/execution-view-store.ts @@ -8,6 +8,35 @@ import type { WorkflowDocument, } from './execution.types'; +/** + * Timestamps come back however the store's driver reports them: the only + * consumer serializes them to JSON, where `Date` and an ISO string are the + * same value, so there is no point parsing or formatting them in between. + */ +export interface ExecutionListItemView { + id: string; + workflowId: string; + status: ExecutionStatus; + mode: ExecutionMode; + createdAt: Date; + updatedAt: Date; + finishedAt: Date | null; +} + +export interface ExecutionListQuery { + workflowIds: string[] | 'all'; + status?: ExecutionStatus[]; + mode?: string; + createdAfter?: string; + createdBefore?: string; + before?: { createdAt: string; id: string }; + /** Required at this layer; the API's default lives on the request schema. */ + limit: number; + includeTotal?: boolean; + /** Same shape the control plane's `ExecutionSummaries.Query` uses. */ + order?: { top?: ExecutionStatus; startedAt?: 'DESC' }; +} + /** * Read view of an execution: what a caller is shown, which is not what running * one needs. It carries the timing the execution path never reads, and omits @@ -66,6 +95,8 @@ export interface ExecutionWithStepsView extends ExecutionView { * columns in the query. */ export interface ExecutionViewStore { + listExecutionViews(query: ExecutionListQuery): Promise; + countExecutionViews(query: ExecutionListQuery): Promise; /** Read view of one execution. Throws `ExecutionNotFoundError` if absent. */ loadExecutionView(id: string): Promise; diff --git a/packages/@n8n/engine/src/index.ts b/packages/@n8n/engine/src/index.ts index 986583fef25..d8f885a85f0 100644 --- a/packages/@n8n/engine/src/index.ts +++ b/packages/@n8n/engine/src/index.ts @@ -16,6 +16,11 @@ export { export type { AuthenticatedCaller, ActionScope, IdentityVerifier } from './auth'; export type { EngineErrorResponse, ExecutionSnapshot, StepDetail } from './server'; +export type { + SearchExecutionsRequest, + SearchExecutionsResponse, + ExecutionListItem, +} 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 b5cf4cf312a..c75d4dea6f8 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 @@ -6,6 +6,7 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } import { AllowAllAdmittance } from '../../admittance'; import { mintIdentityToken, SharedSecretIdentityVerifier } from '../../auth'; +import { UnexpectedError } from '../../common'; import { createDataSource, createStores, WorkflowExecution } from '../../database'; import { generateId } from '../../database/generate-id'; import { ExecutionQueryService, StartExecutionService } from '../../execution'; @@ -13,6 +14,7 @@ import type { WorkflowGraph } from '../../graph'; import type { OrchestrationMessage, WorkQueue } from '../../queue'; import { createEngineRuntime } from '../../runtime'; import { startEngineServer } from '../../testing/start-engine-server'; +import type { SearchExecutionsResponse } from '../api.types'; const sampleGraph: WorkflowGraph = { nodes: [{ id: 'trigger', name: 'Manual Trigger', type: 'trigger', config: {} }], @@ -29,10 +31,140 @@ const sampleWorkflow = { const secret = 'a'.repeat(32); +/** A well-formed cursor, to test what pairs with it rather than its own shape. */ +const sampleCursor = { id: generateId(), createdAt: '2026-09-07T12:00:00.000Z' }; + const authHeader = () => ({ authorization: `Bearer ${mintIdentityToken(secret, { cpId: 'cp-1', tenantId: 'tenant-1' })}`, }); +describe('POST /api/workflow-executions/search (integration)', () => { + const search = (body: object) => + request(url).post('/api/workflow-executions/search').set(authHeader()).send(body); + async function start(workflowId: string) { + const body = startBody({ workflowId }); + await request(url).post('/api/workflow-executions').set(authHeader()).send(body).expect(201); + return body.executionId; + } + + it('requires authentication', async () => { + await request(url) + .post('/api/workflow-executions/search') + .send({ workflowIds: 'all' }) + .expect(401); + }); + + it.each([ + {}, + { workflowIds: [] }, + { workflowIds: [''] }, + { workflowIds: 'all', tenantId: 'tenant' }, + { workflowIds: 'all', unknown: true }, + { workflowIds: 'all', limit: 101 }, + { workflowIds: 'all', limit: 0 }, + { workflowIds: 'all', status: [] }, + { workflowIds: 'all', before: { id: 'bad', createdAt: 'today' } }, + { workflowIds: Array.from({ length: 10_001 }, () => 'wf') }, + { workflowIds: 'all', before: sampleCursor, order: { top: 'running' } }, + ])('rejects an invalid search body %#', async (body) => { + await search(body).expect(400); + }); + + it('rejects a cursor paired with a status-first sort at the store', async () => { + const { executionViewStore } = createStores(dataSource); + await expect( + executionViewStore.listExecutionViews({ + workflowIds: 'all', + limit: 20, + before: sampleCursor, + order: { top: 'running' }, + }), + ).rejects.toThrow(UnexpectedError); + }); + + it('filters workflows and selects exactly the summary columns', async () => { + const workflowId = generateId(); + const id = await start(workflowId); + await start(generateId()); + const response = await search({ workflowIds: [workflowId], includeTotal: true }).expect(200); + const result = response.body as SearchExecutionsResponse; + expect(result.total).toBe(1); + expect(result.nextCursor).toBeNull(); + expect(result.items.map((item) => item.id)).toEqual([id]); + expect(Object.keys(result.items[0]).sort()).toEqual( + ['id', 'workflowId', 'status', 'mode', 'createdAt', 'updatedAt', 'finishedAt'].sort(), + ); + const { executionViewStore } = createStores(dataSource); + const rows = await executionViewStore.listExecutionViews({ + workflowIds: [workflowId], + limit: 20, + }); + expect(Object.keys(rows[0]).sort()).toEqual(Object.keys(result.items[0]).sort()); + }); + + it('allows all for the authenticated CP', async () => { + const id = await start(generateId()); + const response = await search({ workflowIds: 'all', includeTotal: true }).expect(200); + const result = response.body as SearchExecutionsResponse; + expect(result.items.map((item) => item.id)).toContain(id); + expect(result.total).toBeGreaterThanOrEqual(1); + }); + + it('compares modes exactly and applies status filters', async () => { + const workflowId = generateId(); + await start(workflowId); + const production = await search({ + workflowIds: [workflowId], + mode: 'production', + status: ['queued'], + }).expect(200); + expect((production.body as SearchExecutionsResponse).items).toHaveLength(1); + for (const filter of [{ mode: 'webhook' }, { mode: 'manual' }, { status: ['completed'] }]) { + const response = await search({ + workflowIds: [workflowId], + ...filter, + includeTotal: true, + }).expect(200); + expect(response.body).toEqual({ items: [], nextCursor: null, total: 0 }); + } + }); + + it('walks equal timestamps without losing rows and keeps the total independent of the cursor', async () => { + const workflowId = generateId(); + const ids: string[] = []; + for (let i = 0; i < 5; i++) ids.push(await start(workflowId)); + const timestamp = '2026-09-07T12:00:00.000Z'; + await dataSource.query('UPDATE workflow_execution SET created_at = $1 WHERE workflow_id = $2', [ + timestamp, + workflowId, + ]); + const seen: string[] = []; + let before: { id: string; createdAt: string } | undefined; + for (let i = 0; i < 3; i++) { + const response = await search({ + workflowIds: [workflowId], + limit: 2, + includeTotal: true, + before, + createdAfter: timestamp, + createdBefore: timestamp, + }).expect(200); + const page = response.body as SearchExecutionsResponse; + expect(page.total).toBe(5); + expect(page.nextCursor !== null).toBe(i < 2); + seen.push(...page.items.map((item) => item.id)); + before = page.nextCursor ?? undefined; + } + expect(seen).toEqual(ids.sort().reverse()); + expect(new Set(seen).size).toBe(5); + const excluded = await search({ + workflowIds: [workflowId], + createdBefore: '2026-09-07T11:59:59.999Z', + }).expect(200); + expect((excluded.body as SearchExecutionsResponse).items).toEqual([]); + }); +}); + /** The caller always mints the id, so every valid body carries one. */ const startBody = (overrides: Record = {}) => ({ workflowId: 'wf-1', diff --git a/packages/@n8n/engine/src/server/api.types.ts b/packages/@n8n/engine/src/server/api.types.ts index 50920a068fa..9ebaa641316 100644 --- a/packages/@n8n/engine/src/server/api.types.ts +++ b/packages/@n8n/engine/src/server/api.types.ts @@ -6,8 +6,24 @@ import type { StepStatus, WorkflowDocument, } from '../execution'; +import type { ExecutionListQuery } from '../execution/execution-view-store'; import type { WorkflowGraph } from '../graph'; +/** A read-only search. The control plane supplies the visibility decision. */ +export type SearchExecutionsRequest = ExecutionListQuery; + +/** `T` without its `K` fields. */ +type Without = Omit; + +export type ExecutionListItem = Without; + +export interface SearchExecutionsResponse { + items: ExecutionListItem[]; + /** The `before` cursor for the next page, or `null` on the last page. */ + nextCursor: { createdAt: string; id: string } | null; + total?: number; +} + /** * `GET /:id` response. Timestamps go out as ISO strings, since `Date` has no * JSON form. diff --git a/packages/@n8n/engine/src/server/index.ts b/packages/@n8n/engine/src/server/index.ts index a4e339de031..7f8c8a75c99 100644 --- a/packages/@n8n/engine/src/server/index.ts +++ b/packages/@n8n/engine/src/server/index.ts @@ -1,5 +1,10 @@ export { createEngineServer } from './create-engine-server'; export type { EngineServerDeps } from './create-engine-server'; export type { ExecutionSnapshot, StepDetail } from './api.types'; +export type { + SearchExecutionsRequest, + SearchExecutionsResponse, + ExecutionListItem, +} 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 c965e8b4e24..0d8aa30d774 100644 --- a/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts +++ b/packages/@n8n/engine/src/server/routes/workflow-executions.handlers.ts @@ -20,6 +20,52 @@ const ExecutionIdParams = z.object({ id: z.string().uuid() }); */ const GetExecutionQuery = z.object({ includeSteps: z.enum(['true', 'false']).optional() }).strict(); +const datetimeStringWithOffset = () => z.string().datetime({ offset: true }); + +const ExecutionStatusSchema = z.enum(['queued', 'running', 'completed', 'failed', 'cancelled']); + +const SearchExecutionsBody = z + .object({ + workflowIds: z.union([z.literal('all'), z.array(z.string().min(1)).min(1).max(10_000)]), + status: z.array(ExecutionStatusSchema).min(1).optional(), + mode: z.string().min(1).max(32).optional(), + createdAfter: datetimeStringWithOffset().optional(), + createdBefore: datetimeStringWithOffset().optional(), + before: z + .object({ createdAt: datetimeStringWithOffset(), id: z.string().uuid() }) + .strict() + .optional(), + limit: z.number().int().min(1).max(100).default(20), + includeTotal: z.boolean().optional(), + /** Same shape the control plane's `ExecutionSummaries.Query` uses. */ + order: z + .object({ top: ExecutionStatusSchema.optional(), startedAt: z.literal('DESC').optional() }) + .strict() + .optional(), + }) + .strict() + // The cursor only walks its own `(createdAt, id)` order, so a status-first + // sort would drop the rows that sort after the cursor row. Ask for one or the + // other. + .refine((body) => !(body.before && body.order?.top), { + message: 'before cannot be combined with order.top', + path: ['before'], + }); + +export function createSearchExecutionsHandler( + executionQuery: ExecutionQueryService, +): RequestHandler { + return async (req, res) => { + const parsed = SearchExecutionsBody.safeParse(req.body); + if (!parsed.success) { + fail(res, 400, { error: 'invalid_request', details: parsed.error.flatten() }); + return; + } + const result = await executionQuery.searchExecutions(parsed.data); + res.status(200).json(result); + }; +} + /** 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); diff --git a/packages/@n8n/engine/src/server/routes/workflow-executions.ts b/packages/@n8n/engine/src/server/routes/workflow-executions.ts index da09eddcbdf..3ce2dc22508 100644 --- a/packages/@n8n/engine/src/server/routes/workflow-executions.ts +++ b/packages/@n8n/engine/src/server/routes/workflow-executions.ts @@ -2,7 +2,10 @@ import { UUID_V7_PATTERN } from '@n8n/constants'; import { Router, type Router as RouterType } from 'express'; import { z } from 'zod'; -import { createGetExecutionHandler } from './workflow-executions.handlers'; +import { + createGetExecutionHandler, + createSearchExecutionsHandler, +} from './workflow-executions.handlers'; import { AdmittanceRejectedError } from '../../admittance'; import { jsonObjectSchema, jsonValueSchema, UnimplementedError } from '../../common'; import { GraphValidationError, MAX_SLOT_INDEX } from '../../graph'; @@ -62,6 +65,7 @@ const StartExecutionBody = z.object({ export function createWorkflowExecutionsRouter(deps: EngineServerDeps): RouterType { const router = Router(); + router.post('/search', createSearchExecutionsHandler(deps.executionQuery)); router.post('/', async (req, res) => { const parsed = StartExecutionBody.safeParse(req.body);