feat(engine): Add endpoints to list and count executions in the data plane (#38109)

Co-authored-by: Stephen Wright <[email protected]> Co-authored-by: Svetoslav Dekov <[email protected]> Co-authored-by: Claude Fable 5.1 <[email protected]> Co-authored-by: n8n-cat-bot[bot] <283985454+n8n-cat-bot[bot]@users.noreply.github.com> Co-authored-by: n8n-cat-bot[bot] <n8n-cat-bot[bot]@users.noreply.github.com> Co-authored-by: Andreas Fitzek <[email protected]> Co-authored-by: Elias Meire <[email protected]> Co-authored-by: Mutasem Aldmour <[email protected]> Co-authored-by: Riqwan Thamir <[email protected]> Co-authored-by: Cursor <[email protected]> Co-authored-by: Guillaume Jacquart <[email protected]> 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 <[email protected]> Co-authored-by: Raúl Gómez Morales <[email protected]> Co-authored-by: subaru <[email protected]> Co-authored-by: Declan Carroll <[email protected]> Co-authored-by: Charlie Kolb <[email protected]> Co-authored-by: Roman Makowiak <[email protected]> Co-authored-by: Tuukka Kantola <[email protected]>
2026-10-11 22:50:06 +00:00 · 2026-09-16 09:06:07 +00:00
parent aa8cc414bb
commit dac4f63f1e
10 changed files with 350 additions and 1 deletions
@@ -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<WorkflowStepExecution>,
 	) {}

+	async listExecutionViews(query: ExecutionListQuery): Promise<ExecutionListItemView[]> {
+		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<ExecutionListItemView>();
+	}
+
+	async countExecutionViews(query: ExecutionListQuery): Promise<number> {
+		return await this.buildListQuery(query).getCount();
+	}
+
+	private buildListQuery(query: ExecutionListQuery): SelectQueryBuilder<WorkflowExecution> {
+		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<ExecutionView> {
 		const row: ExecutionView | undefined = await this.selectExecution(id).getRawOne();
 		if (!row) throw new ExecutionNotFoundError(id);
@@ -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> = {}): 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<ExecutionViewStore> = {}): 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({
@@ -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<ExecutionView> {
 		return await this.viewStore.loadExecutionView(id);
@@ -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<ExecutionListItemView[]>;
+	countExecutionViews(query: ExecutionListQuery): Promise<number>;
 	/** Read view of one execution. Throws `ExecutionNotFoundError` if absent. */
 	loadExecutionView(id: string): Promise<ExecutionView>;

@@ -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 {
@@ -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<string, unknown> = {}) => ({
 	workflowId: 'wf-1',
@@ -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<T, K extends keyof T> = Omit<T, K>;
+
+export type ExecutionListItem = Without<ExecutionSnapshot, 'graph' | 'workflow' | 'steps'>;
+
+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.
@@ -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';
@@ -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);
@@ -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);