feat(engine): Read executions from the data plane and merge with the DB (#38110)

This commit is contained in:
Tomi Turtiainen
2026-09-16 10:45:58 +00:00
committed by GitHub
parent c5935b9cb5
commit 9b66a2bbfa
36 changed files with 1961 additions and 473 deletions
@@ -0,0 +1,76 @@
import { GlobalConfig } from '@n8n/config';
import { Service } from '@n8n/di';
import { DataSource, In } from '@n8n/typeorm';
import type { SelectQueryBuilder } from '@n8n/typeorm';
import { BaseRepository } from './base-repository';
import { SharedWorkflowRepository } from './shared-workflow.repository';
import { User, WorkflowEntity } from '../entities';
import type { ExecutionSummaries } from '../entities/types-db';
import { TransactionRunner } from '../services/transaction';
import { applyWorkflowBooleanSettingFilter } from '../utils/apply-workflow-boolean-setting-filter';
/** Workflow visibility and attributes for the merged (CP + DP) execution list. */
@Service()
export class ExecutionListRepository extends BaseRepository<WorkflowEntity> {
constructor(
dataSource: DataSource,
private readonly globalConfig: GlobalConfig,
private readonly sharedWorkflowRepository: SharedWorkflowRepository,
transactionRunner: TransactionRunner,
) {
super(WorkflowEntity, dataSource.manager, transactionRunner);
}
/** Resolve the same workflow visibility and attributes as the execution list. */
async findWorkflowIdsForExecutionList(query: ExecutionSummaries.RangeQuery): Promise<string[]> {
if (!query.user || !query.sharingOptions) return [];
const qb = this.createQueryBuilder('workflow').select('workflow.id', 'id');
this.applySharedWithUserFilter(qb, query.user, query.sharingOptions);
this.applyExecutionListFilters(qb, query);
const rows = await qb.distinct(true).getRawMany<{ id: string }>();
return rows.map(({ id }) => id);
}
/** Restricts `qb` to workflows shared with `user` under `sharingOptions`. */
private applySharedWithUserFilter(
qb: SelectQueryBuilder<WorkflowEntity>,
user: User,
sharingOptions: NonNullable<ExecutionSummaries.RangeQuery['sharingOptions']>,
): void {
const shared = this.sharedWorkflowRepository
.buildSharedWorkflowIdsSubquery(user, sharingOptions)
.andWhere('sw.workflowId = workflow.id');
qb.where(`EXISTS (${shared.getQuery()})`).setParameters(shared.getParameters());
}
private applyExecutionListFilters(
qb: SelectQueryBuilder<WorkflowEntity>,
query: ExecutionSummaries.RangeQuery,
): void {
if (query.workflowId) {
qb.andWhere('workflow.id = :workflowId', { workflowId: query.workflowId });
}
if (query.projectId) {
qb.innerJoin('workflow.shared', 'projectShare', 'projectShare.projectId = :projectId', {
projectId: query.projectId,
});
}
if (query.isArchived !== undefined) {
qb.andWhere('workflow.isArchived = :isArchived', { isArchived: query.isArchived });
}
for (const filter of query.workflowBooleanSettings ?? []) {
applyWorkflowBooleanSettingFilter(qb, this.globalConfig, filter.key, filter.value);
}
}
async findNamesForExecutionList(
ids: string[],
): Promise<Array<Pick<WorkflowEntity, 'id' | 'name'>>> {
if (ids.length === 0) return [];
return await this.find({ select: ['id', 'name'], where: { id: In(ids) } });
}
}
@@ -34,6 +34,7 @@ export {
export { EvaluationConfigRepository } from './evaluation-config.repository';
export { ExecutionAnnotationRepository } from './execution-annotation.repository';
export { ExecutionDataRepository } from './execution-data.repository';
export { ExecutionListRepository } from './execution-list.repository';
export { ExecutionMetadataRepository } from './execution-metadata.repository';
export {
ExecutionRepository,
@@ -24,7 +24,7 @@ export interface ExecutionListItemView {
}
export interface ExecutionListQuery {
workflowIds: string[] | 'all';
workflowIds: readonly string[] | 'all';
status?: ExecutionStatus[];
mode?: string;
createdAfter?: string;
@@ -4,7 +4,10 @@ import { mock } from 'vitest-mock-extended';
import type { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
import { EngineV2ExecutionReader } from '../engine-v2-execution-reader.service';
import {
EngineV2ExecutionReader,
type EngineV2SearchQuery,
} from '../engine-v2-execution-reader.service';
import type { ExecutionIdV2 } from '../execution-id';
const EXECUTION_ID = '01a038ae-c4a8-7799-8a3e-e3c2ca055cfa' as ExecutionIdV2;
@@ -53,6 +56,115 @@ describe('EngineV2ExecutionReader', () => {
beforeEach(() => {
vi.clearAllMocks();
dataPlane.getExecution.mockResolvedValue(snapshot());
dataPlane.searchExecutions.mockResolvedValue({ items: [], total: 0, nextCursor: null });
});
describe('findMany', () => {
const query: EngineV2SearchQuery = { limit: 2 };
it.each(['manual', 'webhook', 'trigger'] as const)(
'forwards the mode %s without conversion',
async (mode) => {
await reader.findMany({ ...query, mode }, ['wf-1']);
expect(dataPlane.searchExecutions).toHaveBeenCalledWith(expect.objectContaining({ mode }));
},
);
it('maps supported statuses and drops statuses that the DP cannot match', async () => {
await reader.findMany({ ...query, status: ['success', 'crashed', 'waiting'] }, 'all');
expect(dataPlane.searchExecutions).toHaveBeenCalledWith(
expect.objectContaining({ status: ['completed'] }),
);
dataPlane.searchExecutions.mockClear();
await reader.findMany({ ...query, status: ['crashed', 'waiting'] }, 'all');
expect(dataPlane.searchExecutions).not.toHaveBeenCalled();
});
// Narrowing an unfiltered list to the mapped statuses would hide any status
// the data plane adds later.
it('sends no status filter when the caller asked for none', async () => {
await reader.findMany(query, 'all');
expect(dataPlane.searchExecutions).toHaveBeenCalledWith(
expect.objectContaining({ status: undefined }),
);
});
it('searches once, deduplicating the workflow IDs', async () => {
const ids = Array.from({ length: 3 }, (_, i) => `wf-${i}`);
await reader.findMany(query, [...ids, ids[0]]);
expect(dataPlane.searchExecutions).toHaveBeenCalledTimes(1);
expect(dataPlane.searchExecutions).toHaveBeenCalledWith(
expect.objectContaining({ workflowIds: ids }),
);
});
it('reports the page the data plane returned, without re-sorting it', async () => {
dataPlane.searchExecutions.mockResolvedValue({
items: [snapshot(), snapshot({ id: '01a038ae-c4a8-7799-8a3e-e3c2ca055cfb' })],
total: 8,
nextCursor: { id: EXECUTION_ID, createdAt: '2026-08-25T10:00:00.000Z' },
});
const result = await reader.findMany(query, ['wf-1']);
expect(result).toMatchObject({
total: 8,
hasMore: true,
items: [
{ id: EXECUTION_ID, annotation: { tags: [] } },
{ id: '01a038ae-c4a8-7799-8a3e-e3c2ca055cfb' },
],
});
expect(dataPlane.getExecution).not.toHaveBeenCalled();
});
// Splitting a wider scope across searches can come later; until then the
// request fails rather than silently reporting a partial list.
it('rejects a scope wider than one search takes', async () => {
const ids = Array.from({ length: 10_001 }, (_, i) => `wf-${i}`);
await expect(reader.findMany(query, ids)).rejects.toThrow(
'Cannot search executions across more than 10000 workflows',
);
expect(dataPlane.searchExecutions).not.toHaveBeenCalled();
});
it('accepts a scope right at the cap', async () => {
const ids = Array.from({ length: 10_000 }, (_, i) => `wf-${i}`);
await expect(reader.findMany(query, ids)).resolves.toMatchObject({ items: [] });
expect(dataPlane.searchExecutions).toHaveBeenCalledTimes(1);
});
it('counts only when the caller asks, since a total costs an extra scan', async () => {
await reader.findMany(query, 'all');
expect(dataPlane.searchExecutions).toHaveBeenCalledWith(
expect.objectContaining({ includeTotal: false }),
);
await reader.findMany(query, 'all', { includeTotal: true });
expect(dataPlane.searchExecutions).toHaveBeenLastCalledWith(
expect.objectContaining({ includeTotal: true }),
);
});
it('passes date bounds and the DP source position independently', async () => {
await reader.findMany(
{
...query,
startedBefore: '2026-09-01',
before: { id: EXECUTION_ID, timestamp: '2026-08-30T00:00:00.000Z' },
},
'all',
);
expect(dataPlane.searchExecutions).toHaveBeenCalledWith(
expect.objectContaining({
createdBefore: '2026-09-01T00:00:00.000Z',
before: { id: EXECUTION_ID, createdAt: '2026-08-30T00:00:00.000Z' },
}),
);
});
});
describe('findOne', () => {
@@ -1,76 +1,32 @@
import {
encodeCursorForId,
encodeExecutionCursor,
parseExecutionCursor,
} from '@/executions/execution-cursor';
import { encodeExecutionCursor, parseExecutionCursor } from '@/executions/execution-cursor';
const time = '2026-09-07T12:00:00.000Z';
const V2_EXECUTION_ID = '01a038ae-c4a8-7799-8a3e-e3c2ca055cfa';
describe('parseExecutionCursor', () => {
it('returns no cursor when the caller asks for the first page', () => {
expect(parseExecutionCursor(undefined)).toBeUndefined();
it('returns an empty cursor when the caller asks for the first page', () => {
expect(parseExecutionCursor(undefined)).toEqual({ version: 1 });
});
it('rejects a cursor that carries no position', () => {
const encoded = Buffer.from(JSON.stringify({ version: 1 })).toString('base64url');
expect(() => parseExecutionCursor(encoded)).toThrow('Invalid execution cursor');
it('round trips both positions', () => {
const cursor = {
version: 1 as const,
v1: { id: '10', timestamp: time },
v2: { id: V2_EXECUTION_ID, timestamp: time },
};
expect(parseExecutionCursor(encodeExecutionCursor(cursor))).toEqual(cursor);
});
it('rejects a cursor that carries a non-numeric id', () => {
const encoded = Buffer.from(JSON.stringify({ version: 1, v1: V2_EXECUTION_ID })).toString(
'base64url',
);
expect(() => parseExecutionCursor(encoded)).toThrow('Invalid execution cursor');
});
it('rejects a v2 position that is not a UUID', () => {
const encoded = Buffer.from(JSON.stringify({ version: 1, v2: '123' })).toString('base64url');
expect(() => parseExecutionCursor(encoded)).toThrow('Invalid execution cursor');
});
it('rejects a valid v2 position, which nothing can page by yet', () => {
const encoded = encodeExecutionCursor(V2_EXECUTION_ID);
expect(() => parseExecutionCursor(encoded)).toThrow('Unsupported execution cursor');
});
it('rejects a cursor that carries both positions', () => {
const encoded = Buffer.from(
JSON.stringify({ version: 1, v1: '123', v2: V2_EXECUTION_ID }),
).toString('base64url');
expect(() => parseExecutionCursor(encoded)).toThrow('Invalid execution cursor');
});
});
describe('encodeExecutionCursor', () => {
it('puts an engine 2.0 ID in the v2 position', () => {
const encoded = encodeExecutionCursor(V2_EXECUTION_ID);
const decoded = JSON.parse(Buffer.from(encoded, 'base64url').toString('utf8'));
expect(decoded).toEqual({ version: 1, v2: V2_EXECUTION_ID });
});
it('puts an engine 1.0 ID in the v1 position', () => {
const encoded = encodeExecutionCursor('123');
const decoded = JSON.parse(Buffer.from(encoded, 'base64url').toString('utf8'));
expect(decoded).toEqual({ version: 1, v1: '123' });
});
});
describe('encodeCursorForId', () => {
it('round-trips a v1 execution id through parseExecutionCursor', () => {
const encoded = encodeCursorForId('123');
expect(encoded).not.toBeNull();
expect(parseExecutionCursor(encoded ?? undefined)).toBe('123');
});
it('returns null for an engine 2.0 (v2) execution id', () => {
expect(encodeCursorForId(V2_EXECUTION_ID)).toBeNull();
it.each([
'',
'!',
Buffer.from('{').toString('base64url'),
Buffer.from(JSON.stringify({ version: 2 })).toString('base64url'),
Buffer.from(
JSON.stringify({ version: 1, v1: { id: V2_EXECUTION_ID, timestamp: time } }),
).toString('base64url'),
Buffer.from(JSON.stringify({ version: 1, v1: { id: '10' } })).toString('base64url'),
])('rejects an invalid cursor %s', (cursor) => {
expect(() => parseExecutionCursor(cursor)).toThrow('Invalid execution cursor');
});
});
@@ -1,86 +1,140 @@
import type { DatabaseConfig } from '@n8n/config';
import type { ExecutionRepository } from '@n8n/db';
import type { ExecutionRepository, ExecutionSummaries } from '@n8n/db';
import type { ExecutionSummary } from 'n8n-workflow';
import { mock } from 'vitest-mock-extended';
import { encodeExecutionCursor } from '@/executions/execution-cursor';
import { COMPLETED_STATUSES, ExecutionListV1Service } from '@/executions/execution-list-v1.service';
import { encodeExecutionCursor } from '../execution-cursor';
import { COMPLETED_STATUSES, ExecutionListV1Service } from '../execution-list-v1.service';
describe('ExecutionListV1Service', () => {
const databaseConfig = mock<DatabaseConfig>({ type: 'sqlite' });
const databaseConfig = mock<DatabaseConfig>();
const executionRepository = mock<ExecutionRepository>();
const service = new ExecutionListV1Service(databaseConfig, executionRepository);
const startedAt = new Date('2026-09-07T12:00:00.000Z');
const row = (id: string, status: ExecutionSummary['status'] = 'success') =>
mock<ExecutionSummary>({ id, status, startedAt, createdAt: startedAt });
const query = (
overrides: Partial<ExecutionSummaries.RangeQuery> = {},
): ExecutionSummaries.RangeQuery => ({
kind: 'range',
range: { limit: 2 },
...overrides,
});
/** The `query` above, as the caller narrows it when the two blocks combine. */
const completedPageQuery = (overrides: Partial<ExecutionSummaries.RangeQuery> = {}) =>
query({ order: { startedAt: 'DESC' }, status: COMPLETED_STATUSES, ...overrides });
/** Every `findManyByRangeQuery` call, paired with the statuses it asked for. */
const rangeQueries = () => executionRepository.findManyByRangeQuery.mock.calls.map(([q]) => q);
const currentQuery = () => rangeQueries().find((q) => q.status?.includes('running'));
const completedQuery = () => rangeQueries().find((q) => !q.status?.includes('running'));
beforeEach(() => {
vi.clearAllMocks();
executionRepository.getLiveExecutionRowsOnPostgres.mockResolvedValue(-1);
vi.resetAllMocks();
databaseConfig.type = 'sqlite';
executionRepository.findManyByRangeQuery.mockResolvedValue([]);
executionRepository.fetchCount.mockResolvedValue(0);
});
describe('nextCursor', () => {
it('findPageWithCount returns a cursor for the last row when the page is full', async () => {
executionRepository.findManyByRangeQuery.mockResolvedValue([
mock<ExecutionSummary>({ id: '2' }),
mock<ExecutionSummary>({ id: '1' }),
]);
describe('findCurrentAndCompleted', () => {
it('puts the current block on top and counts the completed page alone', async () => {
executionRepository.findManyByRangeQuery.mockImplementation(async (q) =>
q.status?.includes('running') ? [row('20', 'running')] : [row('10'), row('9')],
);
executionRepository.fetchCount.mockResolvedValue(7);
const { nextCursor } = await service.findPageWithCount(mock({ range: { limit: 2 } }));
const result = await service.findCurrentAndCompleted(completedPageQuery());
expect(nextCursor).toBe(encodeExecutionCursor('1'));
expect(result.results.map((r) => r.id)).toEqual(['20', '10', '9']);
// The count covers the paginated page only, so "current" is excluded.
expect(result.count).toBe(7);
expect(currentQuery()?.order).toEqual({ top: 'running' });
expect(completedQuery()?.status).toEqual(COMPLETED_STATUSES);
});
it('findPageWithCount returns null when the page is partial', async () => {
executionRepository.findManyByRangeQuery.mockResolvedValue([
mock<ExecutionSummary>({ id: '1' }),
]);
it('refetches current in full, ignoring the caller beforeId', async () => {
await service.findCurrentAndCompleted(
completedPageQuery({ range: { limit: 2, beforeId: '99' } }),
);
const { nextCursor } = await service.findPageWithCount(mock({ range: { limit: 20 } }));
expect(currentQuery()?.range).not.toHaveProperty('beforeId');
expect(completedQuery()?.range.beforeId).toBe('99');
});
});
describe('nextCursor', () => {
it('returns a cursor for the last row when the page is full', async () => {
executionRepository.findManyByRangeQuery.mockResolvedValue([row('2'), row('1')]);
const { nextCursor } = await service.findPageWithCount(query({ range: { limit: 2 } }));
expect(nextCursor).toBe(
encodeExecutionCursor({
version: 1,
v1: { id: '1', timestamp: startedAt.toISOString() },
}),
);
});
it('returns null when the page is partial', async () => {
executionRepository.findManyByRangeQuery.mockResolvedValue([row('1')]);
const { nextCursor } = await service.findPageWithCount(query({ range: { limit: 20 } }));
expect(nextCursor).toBeNull();
});
it('findCurrentAndCompleted derives the cursor from the completed page, not current', async () => {
executionRepository.findManyByRangeQuery.mockImplementation(async (query) =>
query.status?.includes('running')
? [mock<ExecutionSummary>({ id: '20' })]
: [mock<ExecutionSummary>({ id: '10' })],
it('derives the cursor from the completed page, not from current', async () => {
executionRepository.findManyByRangeQuery.mockImplementation(async (q) =>
q.status?.includes('running') ? [row('20', 'running')] : [row('10')],
);
const { nextCursor } = await service.findCurrentAndCompleted({
kind: 'range',
status: COMPLETED_STATUSES,
range: { limit: 1 },
const { nextCursor } = await service.findCurrentAndCompleted(
completedPageQuery({ range: { limit: 1 } }),
);
expect(nextCursor).toBe(
encodeExecutionCursor({
version: 1,
v1: { id: '10', timestamp: startedAt.toISOString() },
}),
);
});
});
describe('findPageWithCount', () => {
it('applies the cursor without mutating the caller query', async () => {
const original = query({ range: { limit: 2 } });
await service.findPageWithCount(original, {
version: 1,
v1: { id: '10', timestamp: startedAt.toISOString() },
});
expect(nextCursor).toBe(encodeExecutionCursor('10'));
expect(rangeQueries()[0].range.beforeId).toBe('10');
expect(original.range.beforeId).toBeUndefined();
});
it('findCurrentAndCompleted applies the cursor to completed rows only', async () => {
executionRepository.findManyByRangeQuery.mockResolvedValue([]);
it('counts without the range bounds', async () => {
await service.findPageWithCount(query());
await service.findCurrentAndCompleted({
kind: 'range',
status: COMPLETED_STATUSES,
range: { limit: 20, beforeId: '10' },
});
const queries = executionRepository.findManyByRangeQuery.mock.calls.map(([query]) => query);
const current = queries.find((query) => query.status?.includes('running'));
const completed = queries.find((query) => !query.status?.includes('running'));
expect(current?.range).not.toHaveProperty('beforeId');
expect(completed?.range.beforeId).toBe('10');
expect(executionRepository.fetchCount).toHaveBeenCalledWith(
expect.not.objectContaining({ range: expect.anything() }),
);
});
});
describe('getExecutionsCountForQuery', () => {
it('returns an exact count outside postgres', async () => {
executionRepository.fetchCount.mockResolvedValue(7);
it('reports an exact count outside postgres', async () => {
executionRepository.fetchCount.mockResolvedValue(12);
const result = await service.getExecutionsCountForQuery(mock({ kind: 'count' }));
expect(result).toEqual({ count: 7, estimated: false });
expect(await service.getExecutionsCountForQuery(mock())).toEqual({
count: 12,
estimated: false,
});
expect(executionRepository.getLiveExecutionRowsOnPostgres).not.toHaveBeenCalled();
});
@@ -88,12 +142,21 @@ describe('ExecutionListV1Service', () => {
databaseConfig.type = 'postgresdb';
executionRepository.getLiveExecutionRowsOnPostgres.mockResolvedValue(200_000);
const result = await service.getExecutionsCountForQuery(mock({ kind: 'count' }));
expect(result).toEqual({ count: 200_000, estimated: true });
expect(await service.getExecutionsCountForQuery(mock())).toEqual({
count: 200_000,
estimated: true,
});
expect(executionRepository.fetchCount).not.toHaveBeenCalled();
});
databaseConfig.type = 'sqlite';
it('passes the unknown-count sentinel through', async () => {
databaseConfig.type = 'postgresdb';
executionRepository.getLiveExecutionRowsOnPostgres.mockResolvedValue(-1);
expect(await service.getExecutionsCountForQuery(mock())).toEqual({
count: -1,
estimated: false,
});
});
});
});
@@ -0,0 +1,305 @@
import type { DatabaseConfig } from '@n8n/config';
import type {
ExecutionListRepository,
ExecutionRepository,
ExecutionSummaries,
User,
} from '@n8n/db';
import type { ExecutionListItem } from '@n8n/engine';
import type { ExecutionSummary } from 'n8n-workflow';
import { mock } from 'vitest-mock-extended';
import type { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
import { EngineV2ExecutionReader } from '../engine-v2-execution-reader.service';
import { parseExecutionCursor } from '../execution-cursor';
import { COMPLETED_STATUSES, ExecutionListV1Service } from '../execution-list-v1.service';
import { ExecutionListV2Service } from '../execution-list-v2.service';
describe('ExecutionListV2Service', () => {
const executions = mock<ExecutionRepository>();
const workflows = mock<ExecutionListRepository>();
const dataPlane = mock<EngineDataPlaneProxyService>();
const databaseConfig = mock<DatabaseConfig>();
const reader = new EngineV2ExecutionReader(dataPlane);
// The real v1 service, so the control-plane queries it builds are asserted too.
const v1Service = new ExecutionListV1Service(databaseConfig, executions);
const service = new ExecutionListV2Service(v1Service, executions, workflows, reader, dataPlane);
const member = mock<User>({ id: 'member', role: { slug: 'global:member', scopes: [] } });
const owner = mock<User>({
id: 'owner',
role: { slug: 'global:owner', scopes: [{ slug: 'workflow:read' }] },
});
const query = (
overrides: Partial<ExecutionSummaries.RangeQuery> = {},
): ExecutionSummaries.RangeQuery => ({
kind: 'range',
user: member,
sharingOptions: { projectRoles: ['project:viewer'], workflowRoles: ['workflow:editor'] },
range: { limit: 2 },
status: ['success'],
...overrides,
});
/** The `query` above, as the caller narrows it when the two blocks combine. */
const completedPageQuery = (overrides: Partial<ExecutionSummaries.RangeQuery> = {}) =>
query({ order: { startedAt: 'DESC' }, status: COMPLETED_STATUSES, ...overrides });
const uuid = '01992380-0000-7000-8000-000000000001';
const uuid2 = '01992380-0000-7000-8000-000000000002';
const time = '2026-09-07T12:00:00.000Z';
const v2Item = (
id: string,
status: ExecutionListItem['status'] = 'completed',
): ExecutionListItem =>
({
id,
workflowId: 'wf-a',
status,
mode: 'manual',
createdAt: time,
updatedAt: time,
finishedAt: time,
}) as ExecutionListItem;
const v1Row = (id: string, status: ExecutionSummary['status'] = 'success') =>
mock<ExecutionSummary>({
id,
workflowId: 'wf-a',
status,
mode: 'manual',
createdAt: new Date(time),
startedAt: new Date(time),
});
/** Every data-plane search request this call made. */
const searchRequests = () => dataPlane.searchExecutions.mock.calls.map(([request]) => request);
/** Every control-plane range query this call made. */
const rangeQueries = () => executions.findManyByRangeQuery.mock.calls.map(([q]) => q);
beforeEach(() => {
vi.resetAllMocks();
databaseConfig.type = 'sqlite';
dataPlane.isAvailable.mockReturnValue(true);
dataPlane.searchExecutions.mockResolvedValue({ items: [], nextCursor: null, total: 0 });
workflows.findWorkflowIdsForExecutionList.mockResolvedValue(['wf-a']);
workflows.findNamesForExecutionList.mockResolvedValue([{ id: 'wf-a', name: 'Workflow A' }]);
executions.findManyByRangeQuery.mockResolvedValue([]);
executions.fetchCount.mockResolvedValue(0);
});
describe('scope', () => {
it('sends only the member visibility set and resolves it again on the next request', async () => {
await service.findPageWithCount(query());
expect(dataPlane.searchExecutions).toHaveBeenLastCalledWith(
expect.objectContaining({ workflowIds: ['wf-a'] }),
);
workflows.findWorkflowIdsForExecutionList.mockResolvedValue(['wf-b']);
await service.findPageWithCount(query());
expect(dataPlane.searchExecutions).toHaveBeenLastCalledWith(
expect.objectContaining({ workflowIds: ['wf-b'] }),
);
});
it('uses all for a global reader without workflow restrictions', async () => {
expect(await service.resolveV2Scope(query({ user: owner }))).toBe('all');
expect(workflows.findWorkflowIdsForExecutionList).not.toHaveBeenCalled();
});
it('resolves workflow attributes even for a global reader', async () => {
const filtered = query({ user: owner, projectId: 'shared-project', isArchived: false });
expect(await service.resolveV2Scope(filtered)).toEqual(['wf-a']);
expect(workflows.findWorkflowIdsForExecutionList).toHaveBeenCalledWith(filtered);
});
it('skips the DP for an inaccessible workflow', async () => {
workflows.findWorkflowIdsForExecutionList.mockResolvedValue([]);
await service.findPageWithCount(query({ workflowId: 'denied' }));
expect(dataPlane.searchExecutions).not.toHaveBeenCalled();
});
});
// The DP search has no equivalent for these filters, so the query is answered
// from the control plane alone rather than rejected.
describe('control-plane-only filters', () => {
const cpOnly: Array<Partial<ExecutionSummaries.RangeQuery>> = [
{ metadata: [{ key: 'k', value: 'v', exactMatch: true }] },
{ annotationTags: ['tag'] },
{ vote: 'up' },
{ workflowVersionId: 'version' },
];
it.each(cpOnly)('skips the DP for %s', async (filter) => {
await service.findPageWithCount(query(filter));
expect(dataPlane.searchExecutions).not.toHaveBeenCalled();
});
it.each(cpOnly)('still reports the v1 page for %s', async (filter) => {
executions.findManyByRangeQuery.mockResolvedValue([v1Row('10')]);
executions.fetchCount.mockResolvedValue(4);
const result = await service.findPageWithCount(query(filter));
expect(result.results.map((r) => r.id)).toEqual(['10']);
expect(result.count).toBe(4);
});
});
describe('merging', () => {
it('merges both stores, enriches v2 names, and sums the counts', async () => {
dataPlane.searchExecutions.mockResolvedValue({
items: [v2Item(uuid)],
nextCursor: null,
total: 3,
});
executions.findManyByRangeQuery.mockResolvedValue([v1Row('10')]);
executions.fetchCount.mockResolvedValue(5);
const result = await service.findPageWithCount(query({ range: { limit: 1 } }));
expect(result).toMatchObject({
count: 8,
estimated: false,
results: [{ id: uuid, workflowName: 'Workflow A' }],
});
// Only a v2 row made the page, so the v1 position stays untouched.
expect(parseExecutionCursor(result.nextCursor!).v1).toBeUndefined();
});
it('preserves the unknown-count sentinel instead of adding to it', async () => {
databaseConfig.type = 'postgresdb';
executions.getLiveExecutionRowsOnPostgres.mockResolvedValue(-1);
dataPlane.searchExecutions.mockResolvedValue({ items: [], total: 20, nextCursor: null });
expect(await service.findPageWithCount(query())).toMatchObject({ count: -1 });
});
it('counts without the cursor bounds', async () => {
await service.findPageWithCount(query());
expect(executions.fetchCount).toHaveBeenCalledWith(
expect.not.objectContaining({ range: expect.anything() }),
);
});
it('fails instead of reporting a partial list when the DP fails', async () => {
dataPlane.searchExecutions.mockRejectedValue(new Error('unavailable'));
await expect(service.findPageWithCount(query())).rejects.toThrow('unavailable');
});
it('asks the control plane for one row past the page, to detect more', async () => {
await service.findPageWithCount(query({ range: { limit: 2 } }));
expect(rangeQueries()[0].range.limit).toBe(3);
});
it('bounds the control plane page by the cursor, else by a caller beforeId', async () => {
const cursor = { version: 1 as const, v1: { id: '10', timestamp: time } };
await service.findPageWithCount(query({ range: { limit: 2, beforeId: '99' } }), cursor);
expect(rangeQueries()[0].range.beforeId).toBe('10');
executions.findManyByRangeQuery.mockClear();
await service.findPageWithCount(query({ range: { limit: 2, beforeId: '99' } }));
expect(rangeQueries()[0].range.beforeId).toBe('99');
});
});
describe('findCurrentAndCompleted', () => {
it('searches current outside the completed continuation', async () => {
await service.findCurrentAndCompleted(completedPageQuery());
expect(searchRequests()).toEqual(
expect.arrayContaining([
expect.objectContaining({
status: ['completed', 'failed', 'cancelled'],
includeTotal: true,
}),
expect.objectContaining({
status: ['queued', 'running'],
includeTotal: false,
order: { top: 'running' },
}),
]),
);
});
it('puts running executions at the top of the current block', async () => {
executions.findManyByRangeQuery.mockImplementation(async (q) =>
q.status?.includes('running') ? [v1Row('20', 'new')] : [],
);
dataPlane.searchExecutions.mockImplementation(async (request) =>
request.status?.includes('running')
? { items: [v2Item(uuid, 'running')], nextCursor: null, total: 0 }
: { items: [], nextCursor: null, total: 0 },
);
const result = await service.findCurrentAndCompleted(completedPageQuery());
expect(result.results.map((r) => r.status)).toEqual(['running', 'new']);
});
it('lets the current block exceed the limit, since each store contributes up to it', async () => {
executions.findManyByRangeQuery.mockImplementation(async (q) =>
q.status?.includes('running') ? [v1Row('20', 'running'), v1Row('19', 'running')] : [],
);
dataPlane.searchExecutions.mockImplementation(async (request) =>
request.status?.includes('running')
? {
items: [v2Item(uuid, 'running'), v2Item(uuid2, 'running')],
nextCursor: null,
total: 0,
}
: { items: [], nextCursor: null, total: 0 },
);
const result = await service.findCurrentAndCompleted(
completedPageQuery({ range: { limit: 2 } }),
);
expect(result.results).toHaveLength(4);
});
});
describe('findPageWithCount', () => {
it('never combines the blocks, so a status-less query pages every status', async () => {
await service.findPageWithCount(query({ status: undefined }));
expect(rangeQueries()).toHaveLength(1);
expect(rangeQueries()[0].status).toBeUndefined();
expect(searchRequests()).toHaveLength(1);
expect(searchRequests()[0].status).toBeUndefined();
});
it('resolves v2 workflow names', async () => {
dataPlane.searchExecutions.mockResolvedValue({
items: [v2Item(uuid)],
nextCursor: null,
total: 1,
});
const result = await service.findPageWithCount(query());
expect(result.results).toEqual([expect.objectContaining({ workflowName: 'Workflow A' })]);
});
});
describe('scope wider than one data plane search', () => {
it('fails rather than reporting a list missing engine 2.0 rows', async () => {
workflows.findWorkflowIdsForExecutionList.mockResolvedValue(
Array.from({ length: 10_001 }, (_, i) => `wf-${i}`),
);
await expect(service.findPageWithCount(query())).rejects.toThrow(
'Cannot search executions across more than 10000 workflows',
);
});
});
});
@@ -1,92 +1,178 @@
import type { ExecutionSummaries } from '@n8n/db';
import type { ExecutionSummaries, User } from '@n8n/db';
import { Container } from '@n8n/di';
import { mock } from 'vitest-mock-extended';
import {
COMPLETED_STATUSES,
type ExecutionListV1Service,
} from '@/executions/execution-list-v1.service';
import { ExecutionListService } from '@/executions/execution-list.service';
import type { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
import type { RoleService } from '@/services/role.service';
import type { WorkflowSharingService } from '@/workflows/workflow-sharing.service';
import type { ExecutionCursor } from '../execution-cursor';
import { COMPLETED_STATUSES, type ExecutionListV1Service } from '../execution-list-v1.service';
import { ExecutionListService } from '../execution-list.service';
import { ExecutionListV2Service } from '../execution-list-v2.service';
describe('ExecutionListService', () => {
const v1Service = mock<ExecutionListV1Service>();
const v2Service = mock<ExecutionListV2Service>();
const dataPlane = mock<EngineDataPlaneProxyService>();
const workflowSharingService = mock<WorkflowSharingService>();
const roleService = mock<RoleService>();
const service = new ExecutionListService(
v1Service,
dataPlane,
workflowSharingService,
roleService,
);
const service = new ExecutionListService(v1Service, workflowSharingService, roleService);
const query = (
overrides: Partial<ExecutionSummaries.RangeQuery> = {},
): ExecutionSummaries.RangeQuery => ({
kind: 'range',
range: { limit: 20 },
...overrides,
});
/** The `query` above, as `listExecutionsForUI` narrows it when the two blocks combine. */
const completedPageQuery = (overrides: Partial<ExecutionSummaries.RangeQuery> = {}) => ({
...query(overrides),
order: { startedAt: 'DESC' },
status: COMPLETED_STATUSES,
});
const cursor: ExecutionCursor = {
version: 1,
v1: { id: '10', timestamp: '2026-09-07T12:00:00.000Z' },
v2: {
id: '01992380-0000-7000-8000-000000000001',
timestamp: '2026-09-07T12:00:00.000Z',
},
};
const page = { results: [], nextCursor: null, count: 0, estimated: false };
beforeEach(() => {
vi.clearAllMocks();
vi.resetAllMocks();
Container.set(ExecutionListV2Service, v2Service);
dataPlane.isAvailable.mockReturnValue(true);
v1Service.findCurrentAndCompleted.mockResolvedValue(page);
v1Service.findPageWithCount.mockResolvedValue(page);
v2Service.findCurrentAndCompleted.mockResolvedValue(page);
v2Service.findPageWithCount.mockResolvedValue(page);
});
describe('provider routing', () => {
it('routes to v1 alone while the data plane is down', async () => {
dataPlane.isAvailable.mockReturnValue(false);
await service.findPageWithCount(query(), cursor);
expect(v1Service.findPageWithCount).toHaveBeenCalledWith(query(), cursor);
expect(v2Service.findPageWithCount).not.toHaveBeenCalled();
});
it('routes to the merged provider while the data plane is up', async () => {
await service.findPageWithCount(query(), cursor);
expect(v2Service.findPageWithCount).toHaveBeenCalledWith(query(), cursor);
expect(v1Service.findPageWithCount).not.toHaveBeenCalled();
});
it('reads availability per request, so a late data plane is picked up', async () => {
dataPlane.isAvailable.mockReturnValueOnce(false).mockReturnValueOnce(true);
await service.findPageWithCount(query());
await service.findPageWithCount(query());
expect(v1Service.findPageWithCount).toHaveBeenCalledTimes(1);
expect(v2Service.findPageWithCount).toHaveBeenCalledTimes(1);
});
});
describe('listExecutionsForUI', () => {
const QUERIES_WITH_STATUS_OR_CURSOR: ExecutionSummaries.RangeQuery[] = [
{ kind: 'range', status: ['waiting'], range: { limit: 20 } },
{ kind: 'range', status: ['waiting'], range: { limit: 20, beforeId: '999' } },
{ kind: 'range', status: undefined, range: { limit: 20, beforeId: '999' } },
{ kind: 'range', status: [], range: { limit: 20, beforeId: '999' } },
];
it('prepends the current block on the first page of a status-less list', async () => {
await service.listExecutionsForUI(query());
const QUERIES_WITHOUT_STATUS_OR_CURSOR: ExecutionSummaries.RangeQuery[] = [
{ kind: 'range', status: undefined, range: { limit: 20 } },
{ kind: 'range', status: [], range: { limit: 20 } },
];
expect(v2Service.findCurrentAndCompleted).toHaveBeenCalledWith(completedPageQuery());
expect(v2Service.findPageWithCount).not.toHaveBeenCalled();
});
test.each(QUERIES_WITH_STATUS_OR_CURSOR)(
'pages a single block when a status filter or a cursor is given',
async (query) => {
await service.listExecutionsForUI(query);
it('keeps the completed filter on later pages, so "load more" never repeats current', async () => {
await service.listExecutionsForUI(query(), cursor);
expect(v1Service.findPageWithCount).toHaveBeenCalledWith(query);
expect(v1Service.findCurrentAndCompleted).not.toHaveBeenCalled();
},
);
expect(v2Service.findPageWithCount).toHaveBeenCalledWith(completedPageQuery(), cursor);
expect(v2Service.findCurrentAndCompleted).not.toHaveBeenCalled();
});
test.each(QUERIES_WITHOUT_STATUS_OR_CURSOR)(
'combines the current and completed blocks on the first page',
async (query) => {
await service.listExecutionsForUI(query);
it('treats a cursor holding no position as a first page', async () => {
await service.listExecutionsForUI(query(), { version: 1 });
expect(v1Service.findCurrentAndCompleted).toHaveBeenCalledWith({
...query,
status: COMPLETED_STATUSES,
});
expect(v1Service.findPageWithCount).not.toHaveBeenCalled();
},
);
expect(v2Service.findCurrentAndCompleted).toHaveBeenCalledWith(completedPageQuery());
});
it('honours an explicit status filter instead of combining current and completed', async () => {
await service.listExecutionsForUI(query({ status: ['error'] }));
expect(v2Service.findPageWithCount).toHaveBeenCalledWith(
{ ...query({ status: ['error'] }), order: { startedAt: 'DESC' } },
undefined,
);
expect(v2Service.findCurrentAndCompleted).not.toHaveBeenCalled();
});
it('orders every page by startedAt, so the merge and the cursor agree', async () => {
await service.listExecutionsForUI(query({ status: ['error'] }), cursor);
expect(v2Service.findPageWithCount.mock.calls[0][0].order).toEqual({ startedAt: 'DESC' });
});
});
describe('addScopes', () => {
it('assigns each summary the scopes of its workflow', async () => {
workflowSharingService.getSharedWorkflowScopes.mockResolvedValue([
['wf-1', ['workflow:read']],
]);
const summaries = [
mock<ExecutionSummaries.ExecutionSummaryWithScopes>({ workflowId: 'wf-1' }),
mock<ExecutionSummaries.ExecutionSummaryWithScopes>({ workflowId: 'wf-2' }),
];
describe('findPageWithCount', () => {
it('never combines nor reorders, so the caller query passes through', async () => {
await service.findPageWithCount(query());
await service.addScopes(mock(), summaries);
expect(summaries[0].scopes).toEqual(['workflow:read']);
expect(summaries[1].scopes).toEqual([]);
expect(v2Service.findPageWithCount).toHaveBeenCalledWith(query(), undefined);
expect(v2Service.findCurrentAndCompleted).not.toHaveBeenCalled();
});
});
describe('buildSharingOptions', () => {
it('resolves the roles that carry the scope', async () => {
roleService.rolesWithScope
.mockResolvedValueOnce(['project:admin'])
.mockResolvedValueOnce(['workflow:owner']);
it('resolves visibility from the role scopes, not from the sharing license', async () => {
roleService.rolesWithScope.mockResolvedValueOnce(['project:viewer']);
roleService.rolesWithScope.mockResolvedValueOnce(['workflow:editor']);
const options = await service.buildSharingOptions('workflow:read');
expect(options).toEqual({
expect(await service.buildSharingOptions('workflow:read')).toEqual({
scopes: ['workflow:read'],
projectRoles: ['project:admin'],
workflowRoles: ['workflow:owner'],
projectRoles: ['project:viewer'],
workflowRoles: ['workflow:editor'],
});
expect(roleService.rolesWithScope).toHaveBeenCalledWith('project', ['workflow:read']);
expect(roleService.rolesWithScope).toHaveBeenCalledWith('workflow', ['workflow:read']);
});
});
describe('addScopes', () => {
it('assigns the workflow scopes onto each summary, deduplicating the lookup', async () => {
workflowSharingService.getSharedWorkflowScopes.mockResolvedValue([
['wf-a', ['workflow:read', 'workflow:execute']],
]);
const summaries = [
{ id: '1', workflowId: 'wf-a' },
{ id: '2', workflowId: 'wf-a' },
{ id: '3', workflowId: 'wf-b' },
] as ExecutionSummaries.ExecutionSummaryWithScopes[];
const user = mock<User>({ id: 'user-1' });
await service.addScopes(user, summaries);
expect(workflowSharingService.getSharedWorkflowScopes).toHaveBeenCalledWith(
['wf-a', 'wf-b'],
user,
);
// A workflow the lookup did not report reads as no scopes, never undefined.
expect(summaries.map((s) => s.scopes)).toEqual([
['workflow:read', 'workflow:execute'],
['workflow:read', 'workflow:execute'],
[],
]);
});
});
});
@@ -11,13 +11,13 @@ import type {
Project,
User,
WorkflowHistoryRepository,
WorkflowHistory,
} from '@n8n/db';
import type { WorkflowHistory } from '@n8n/db';
import { Container } from '@n8n/di';
import { QueryFailedError } from '@n8n/typeorm';
import { mock } from 'vitest-mock-extended';
import type { IRun, IRunData, IRunExecutionData, ITaskData } from 'n8n-workflow';
import { ManualExecutionCancelledError, WorkflowOperationError } from 'n8n-workflow';
import { mock } from 'vitest-mock-extended';
import type { ActiveExecutions } from '@/active-executions';
import type { ConcurrencyControlService } from '@/concurrency/concurrency-control.service';
@@ -25,13 +25,13 @@ import { AbortedExecutionRetryError } from '@/errors/aborted-execution-retry.err
import { MissingExecutionStopError } from '@/errors/missing-execution-stop.error';
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
import { NotFoundError } from '@/errors/response-errors/not-found.error';
import type { EventService } from '@/events/event.service';
import type { EngineV2ExecutionReader } from '@/executions/engine-v2-execution-reader.service';
import { MissingExecutionDataError } from '@/executions/execution-data/missing-execution-data.error';
import type { ExecutionPersistence } from '@/executions/execution-persistence';
import type { EngineV2ExecutionReader } from '@/executions/engine-v2-execution-reader.service';
import type { ExecutionRedactionServiceProxy } from '@/executions/execution-redaction-proxy.service';
import { ExecutionService } from '@/executions/execution.service';
import type { ExecutionRequest } from '@/executions/execution.types';
import type { EventService } from '@/events/event.service';
import type { ExecutionStopService } from '@/scaling/execution-stop.service';
import { ScalingService } from '@/scaling/scaling.service';
import type { Job } from '@/scaling/scaling.types';
@@ -5,7 +5,7 @@ import { mock } from 'vitest-mock-extended';
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
import { NotFoundError } from '@/errors/response-errors/not-found.error';
import { NotImplementedError } from '@/errors/response-errors/not-implemented.error';
import type { License } from '@/license';
import type { ExecutionCursor } from '@/executions/execution-cursor';
import type { ExecutionListService } from '@/executions/execution-list.service';
import type { ExecutionService } from '@/executions/execution.service';
import type { ExecutionRequest } from '@/executions/execution.types';
@@ -18,14 +18,13 @@ describe('ExecutionsController', () => {
const executionService = mock<ExecutionService>();
const executionListService = mock<ExecutionListService>();
const workflowSharingService = mock<WorkflowSharingService>();
const license = mock<License>();
const executionsController = new ExecutionsController(
executionService,
executionListService,
mock(),
workflowSharingService,
license,
mock(),
executionListService,
);
beforeEach(() => {
@@ -96,55 +95,46 @@ describe('ExecutionsController', () => {
});
describe('getMany', () => {
const NO_EXECUTIONS = {
count: 0,
estimated: false,
results: [],
nextCursor: null,
};
it.each([undefined, [], ['success']] as const)(
'passes filters and cursor to the editor list for %s',
async (status) => {
const rangeQuery: ExecutionSummaries.RangeQuery = {
kind: 'range',
range: { limit: 20 },
status: status ? [...status] : undefined,
};
const user = mock<User>({ id: 'member' });
const sharingOptions = {
scopes: ['workflow:read' as const],
workflowRoles: ['workflow:editor'],
projectRoles: ['project:viewer'],
};
executionListService.buildSharingOptions.mockResolvedValue(sharingOptions);
executionListService.listExecutionsForUI.mockResolvedValue({
results: [],
count: 0,
estimated: false,
nextCursor: null,
});
// The middleware decodes the cursor, so the controller forwards the
// decoded object rather than the query param.
const cursor: ExecutionCursor = {
version: 1,
v1: { id: '10', timestamp: '2026-09-07T12:00:00.000Z' },
};
const req = mock<ExecutionRequest.GetMany>({ rangeQuery, user, cursor });
it('should list executions for the UI and add scopes', async () => {
const rangeQuery: ExecutionSummaries.RangeQuery = {
kind: 'range',
workflowId: undefined,
status: undefined,
range: { limit: 20, beforeId: undefined },
};
const sharingOptions = {
scopes: ['workflow:read' as const],
workflowRoles: [],
projectRoles: [],
};
executionListService.buildSharingOptions.mockResolvedValue(sharingOptions);
executionListService.listExecutionsForUI.mockResolvedValue(NO_EXECUTIONS);
executionService.getConcurrentExecutionsCount.mockResolvedValue(-1);
const req = mock<ExecutionRequest.GetMany>({ rangeQuery });
const result = await executionsController.getMany(req);
expect(rangeQuery.sharingOptions).toEqual(sharingOptions);
expect(executionListService.listExecutionsForUI).toHaveBeenCalledWith(rangeQuery);
expect(executionListService.addScopes).toHaveBeenCalledWith(req.user, NO_EXECUTIONS.results);
expect(executionService.getConcurrentExecutionsCount).toHaveBeenCalled();
expect(result).toEqual({ ...NO_EXECUTIONS, concurrentExecutionsCount: -1 });
});
it('should drop the advanced filters without the license', async () => {
const rangeQuery: ExecutionSummaries.RangeQuery = {
kind: 'range',
range: { limit: 20 },
metadata: [{ key: 'k', value: 'v' }],
annotationTags: ['tag-1'],
};
license.isAdvancedExecutionFiltersEnabled.mockReturnValue(false);
executionListService.listExecutionsForUI.mockResolvedValue(NO_EXECUTIONS);
await executionsController.getMany(mock<ExecutionRequest.GetMany>({ rangeQuery }));
expect(rangeQuery.metadata).toBeUndefined();
expect(rangeQuery.annotationTags).toBeUndefined();
});
await expect(executionsController.getMany(req)).resolves.toMatchObject({
results: [],
nextCursor: null,
});
expect(executionListService.listExecutionsForUI).toHaveBeenCalledWith(
expect.objectContaining({ user, sharingOptions, status: rangeQuery.status }),
cursor,
);
expect(executionListService.addScopes).toHaveBeenCalledWith(user, []);
},
);
});
describe('stop', () => {
@@ -0,0 +1,60 @@
import type { ExecutionSummary } from 'n8n-workflow';
import { parseExecutionCursor } from '../execution-cursor';
import { mergeExecutionPages } from '../merge-execution-pages';
const time = '2026-09-07T12:00:00.000Z';
const uuid = '01992380-0000-7000-8000-000000000001';
const item = (id: string, timestamp = time): ExecutionSummary => ({
id,
workflowId: 'wf',
mode: 'manual',
status: 'success',
createdAt: new Date(timestamp),
startedAt: new Date(timestamp),
});
describe('mergeExecutionPages', () => {
it('walks tied source pages without advancing an unconsumed source', () => {
const first = mergeExecutionPages(
[
{ items: [item('10'), item('9')], hasMore: false },
{ items: [item(uuid)], hasMore: false },
],
1,
{ version: 1 },
);
expect(first.results.map((row) => row.id)).toEqual([uuid]);
const cursor = parseExecutionCursor(first.nextCursor!);
expect(cursor.v1).toBeUndefined();
expect(cursor.v2).toEqual({ id: uuid, timestamp: time });
const second = mergeExecutionPages(
[{ items: [item('10'), item('9')], hasMore: false }],
1,
cursor,
);
expect(second.results.map((row) => row.id)).toEqual(['10']);
expect(parseExecutionCursor(second.nextCursor!).v2).toEqual(cursor.v2);
const third = mergeExecutionPages(
[{ items: [item('9')], hasMore: false }],
1,
parseExecutionCursor(second.nextCursor!),
);
expect(third.results.map((row) => row.id)).toEqual(['9']);
expect(third.nextCursor).toBeNull();
});
it('uses creation time for a row with no start time', () => {
const row = { ...item('1', '2026-09-07T12:00:01.000Z'), startedAt: null };
expect(
mergeExecutionPages([{ items: [item(uuid), row], hasMore: false }], 2, { version: 1 })
.results,
).toEqual([row, item(uuid)]);
});
it('keeps a continuation when the source has unseen rows', () => {
expect(
mergeExecutionPages([{ items: [item('1')], hasMore: true }], 1, { version: 1 }).nextCursor,
).not.toBeNull();
});
});
@@ -1,4 +1,3 @@
import type { SerializedCursor } from '@n8n/api-types';
import type { NextFunction } from 'express';
import type * as express from 'express';
import { mock } from 'vitest-mock-extended';
@@ -14,6 +13,24 @@ describe('`parseRangeQuery` middleware', () => {
const nextFn: NextFunction = vi.fn();
/**
* The middleware reads the raw express query, which still carries the legacy
* `firstId` / `lastId` params that the request type no longer declares. Every
* param it reads needs an explicit `undefined`, or `mock` auto-mocks the key
* into a function and the middleware sees a value that was never sent.
*/
const request = (query: Record<string, string | undefined>) =>
mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: undefined,
limit: undefined,
firstId: undefined,
lastId: undefined,
...query,
} as unknown as ExecutionRequest.GetMany['query'],
});
beforeEach(() => {
vi.restoreAllMocks();
vi.clearAllMocks();
@@ -21,55 +38,27 @@ describe('`parseRangeQuery` middleware', () => {
describe('errors', () => {
test('should fail on invalid JSON', () => {
const statusSpy = res.status;
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: '{ "status": ["waiting }',
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
const req = request({ filter: '{ "status": ["waiting }' });
parseRangeQuery(req, res, nextFn);
expect(nextFn).toBeCalledTimes(0);
expect(statusSpy).toBeCalledWith(400);
expect(res.status).toBeCalledWith(400);
});
test('should fail on invalid schema', () => {
const statusSpy = res.status;
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: '{ "status": 123 }',
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
const req = request({ filter: '{ "status": 123 }' });
parseRangeQuery(req, res, nextFn);
expect(nextFn).toBeCalledTimes(0);
expect(statusSpy).toBeCalledWith(400);
expect(res.status).toBeCalledWith(400);
});
});
describe('filter', () => {
test('should parse status and mode fields', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: '{ "status": ["waiting"], "mode": "manual" }',
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
const req = request({ filter: '{ "status": ["waiting"], "mode": "manual" }' });
parseRangeQuery(req, res, nextFn);
@@ -79,14 +68,8 @@ describe('`parseRangeQuery` middleware', () => {
});
test('should parse date-related fields', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: '{ "startedBefore": "2021-01-01", "startedAfter": "2020-01-01" }',
limit: undefined,
firstId: undefined,
lastId: undefined,
},
const req = request({
filter: '{ "startedBefore": "2021-01-01", "startedAfter": "2020-01-01" }',
});
parseRangeQuery(req, res, nextFn);
@@ -97,15 +80,7 @@ describe('`parseRangeQuery` middleware', () => {
});
test('should parse `workflowId` field', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: '{ "workflowId": "456" }',
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
const req = request({ filter: '{ "workflowId": "456" }' });
parseRangeQuery(req, res, nextFn);
@@ -114,15 +89,7 @@ describe('`parseRangeQuery` middleware', () => {
});
test('should parse `projectId` field', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: '{ "projectId": "123" }',
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
const req = request({ filter: '{ "projectId": "123" }' });
parseRangeQuery(req, res, nextFn);
@@ -131,16 +98,8 @@ describe('`parseRangeQuery` middleware', () => {
});
test('should delete invalid fields', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
// `id` is no longer a filter, so it is dropped like any unknown field.
filter: '{ "workflowId": "456", "id": "123", "test": "789" }',
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
// `id` is no longer a filter, so it is dropped like any unknown field.
const req = request({ filter: '{ "workflowId": "456", "id": "123", "test": "789" }' });
parseRangeQuery(req, res, nextFn);
@@ -152,16 +111,8 @@ describe('`parseRangeQuery` middleware', () => {
});
describe('range', () => {
test('should reject first and last IDs', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: undefined,
limit: undefined,
firstId: '111',
lastId: '999',
},
});
test.each(['firstId', 'lastId'])('should reject the legacy `%s` param', (param) => {
const req = request({ [param]: '111' });
parseRangeQuery(req, res, nextFn);
@@ -170,15 +121,7 @@ describe('`parseRangeQuery` middleware', () => {
});
test('should parse limit', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: undefined,
limit: '50',
firstId: undefined,
lastId: undefined,
},
});
const req = request({ limit: '50' });
parseRangeQuery(req, res, nextFn);
@@ -187,15 +130,7 @@ describe('`parseRangeQuery` middleware', () => {
});
test('should default limit to 20 if absent', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: undefined,
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
const req = request({});
parseRangeQuery(req, res, nextFn);
@@ -203,51 +138,46 @@ describe('`parseRangeQuery` middleware', () => {
expect(nextFn).toBeCalledTimes(1);
});
test('should leave `beforeId` unset without a cursor', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: undefined,
filter: undefined,
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
test('should leave the cursor unset without a `cursor` param', () => {
const req = request({});
parseRangeQuery(req, res, nextFn);
expect(req.cursor).toBeUndefined();
expect(req.rangeQuery.range.beforeId).toBeUndefined();
expect(nextFn).toBeCalledTimes(1);
});
test('should set range.beforeId from a valid cursor', () => {
const cursor = encodeExecutionCursor('123');
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor,
filter: undefined,
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
// The middleware only decodes the cursor. Each store derives its own page
// bound from it, so `range.beforeId` stays unset here.
test('should decode a v1 cursor position', () => {
const v1 = { id: '123', timestamp: '2026-09-07T12:00:00.000Z' };
const req = request({ cursor: encodeExecutionCursor({ version: 1, v1 }) });
parseRangeQuery(req, res, nextFn);
expect(req.rangeQuery.range.beforeId).toBe('123');
expect(req.cursor?.v1).toEqual(v1);
expect(req.rangeQuery.range.beforeId).toBeUndefined();
expect(nextFn).toBeCalledTimes(1);
});
test('should reject an invalid cursor', () => {
const req = mock<ExecutionRequest.GetMany>({
query: {
cursor: 'not-a-real-cursor' as SerializedCursor,
filter: undefined,
limit: undefined,
firstId: undefined,
lastId: undefined,
},
});
test('should decode a v2 cursor position', () => {
const v2 = {
id: '01a038ae-c4a8-7799-8a3e-e3c2ca055cfa',
timestamp: '2026-09-07T12:00:00.000Z',
};
const req = request({ cursor: encodeExecutionCursor({ version: 1, v2 }) });
parseRangeQuery(req, res, nextFn);
expect(req.cursor?.v2).toEqual(v2);
expect(req.cursor?.v1).toBeUndefined();
expect(req.rangeQuery.range.beforeId).toBeUndefined();
expect(nextFn).toBeCalledTimes(1);
});
test.each(['not-a-real-cursor', ''])('should reject the invalid cursor %j', (cursor) => {
const req = request({ cursor });
parseRangeQuery(req, res, nextFn);
@@ -1,31 +1,54 @@
import type { IExecutionResponse } from '@n8n/db';
import { Service } from '@n8n/di';
import type { ExecutionMode, ExecutionSnapshot, ExecutionStatus } from '@n8n/engine';
import type {
ExecutionListItem,
ExecutionSnapshot,
ExecutionStatus,
SearchExecutionsRequest,
} from '@n8n/engine';
import type {
ExecutionStatus as ExecutionStatusV1,
IRunExecutionData,
WorkflowExecuteMode,
ExecutionSummary,
Workflow,
} from 'n8n-workflow';
import { UserError } from 'n8n-workflow';
import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
import { resolveV2Statuses, toV1Mode, toV1Status } from './engine-v2-mapping';
import type { ExecutionPosition } from './execution-cursor';
import { toWorkflowSnapshot, type WorkflowSnapshot } from './execution-data/types';
import type { ExecutionIdV2 } from './execution-id';
/** A status added later reads as `unknown` rather than being guessed at. */
const STATUS_V1 = new Map<ExecutionStatus, ExecutionStatusV1>([
['queued', 'new'],
['running', 'running'],
['completed', 'success'],
['failed', 'error'],
['cancelled', 'canceled'],
]);
/**
* Workflow visibility scope for engine v2. Marks which workflows can be
* returned by the search.
*/
export type V2Scope = 'all' | ReadonlyArray<Workflow['id']>;
/** Anything not manual is a production run. */
const MODE_V1 = new Map<ExecutionMode, WorkflowExecuteMode>([
['manual', 'manual'],
['production', 'trigger'],
]);
/** The subset of a range query the data plane search actually understands. */
export interface EngineV2SearchQuery {
/** A v1 status filter; resolved internally to the v2 statuses that map onto it. */
status?: ExecutionStatusV1[];
mode?: WorkflowExecuteMode;
startedAfter?: string;
startedBefore?: string;
/** The DP source position to resume from, independent of the control plane's own cursor. */
before?: ExecutionPosition;
limit: number;
order?: {
top?: 'queued' | 'running' | 'completed' | 'failed' | 'cancelled';
startedAt?: 'DESC';
};
}
/**
* The most workflow IDs one data plane search takes. A wider scope throws for
* now; splitting it across several searches can come later.
*/
const MAX_SCOPE_SIZE = 10_000;
/**
* Reads an engine 2.0 execution for display. The data plane is its only store:
@@ -36,6 +59,84 @@ const MODE_V1 = new Map<ExecutionMode, WorkflowExecuteMode>([
export class EngineV2ExecutionReader {
constructor(private readonly dataPlane: EngineDataPlaneProxyService) {}
async findMany(
query: EngineV2SearchQuery,
scope: V2Scope,
options: { includeTotal?: boolean } = {},
) {
const { includeTotal = false } = options;
if (scope !== 'all' && !scope.length) return this.empty();
const status = resolveV2Statuses(query.status);
// A filter that matches no v2 status can only return nothing.
if (status?.length === 0) return this.empty();
const request = this.buildSearchRequest(
query,
this.searchableScope(scope),
status,
includeTotal,
);
const page = await this.dataPlane.searchExecutions(request);
return {
items: page.items.map((item) => this.toExecutionSummary(item)),
total: page.total ?? 0,
hasMore: page.nextCursor !== null,
};
}
private empty() {
return { items: [], hasMore: false, total: 0 };
}
/** The deduplicated scope, or a user-facing error when one search cannot hold it. */
private searchableScope(scope: V2Scope): V2Scope {
if (scope === 'all') return scope;
const ids = [...new Set(scope)];
if (ids.length > MAX_SCOPE_SIZE) {
throw new UserError(
`Cannot search executions across more than ${MAX_SCOPE_SIZE} workflows. Filter by project or workflow.`,
);
}
return ids;
}
private buildSearchRequest(
query: EngineV2SearchQuery,
scope: V2Scope,
status: ExecutionStatus[] | undefined,
includeTotal: boolean,
): SearchExecutionsRequest {
return {
workflowIds: scope,
status,
mode: query.mode,
createdAfter: query.startedAfter ? new Date(query.startedAfter).toISOString() : undefined,
createdBefore: query.startedBefore ? new Date(query.startedBefore).toISOString() : undefined,
before: query.before ? { createdAt: query.before.timestamp, id: query.before.id } : undefined,
limit: query.limit,
includeTotal,
order: query.order,
};
}
private toExecutionSummary(item: ExecutionListItem): ExecutionSummary {
return {
id: item.id,
workflowId: item.workflowId,
status: toV1Status(item.status),
mode: toV1Mode(item.mode),
finished: item.status === 'completed',
createdAt: new Date(item.createdAt),
startedAt: new Date(item.createdAt),
stoppedAt: item.finishedAt ? new Date(item.finishedAt) : undefined,
annotation: { tags: [] },
};
}
/** `undefined` for absent and for inaccessible alike, so neither reveals the other. */
async findOne(
executionId: ExecutionIdV2,
@@ -74,8 +175,8 @@ export class EngineV2ExecutionReader {
return {
id: snapshot.id,
workflowId: snapshot.workflowId,
mode: MODE_V1.get(snapshot.mode) ?? 'trigger',
status: STATUS_V1.get(snapshot.status) ?? 'unknown',
mode: toV1Mode(snapshot.mode),
status: toV1Status(snapshot.status),
finished: snapshot.status === 'completed',
createdAt: startedAt,
startedAt,
@@ -0,0 +1,38 @@
import type { ExecutionMode, ExecutionStatus } from '@n8n/engine';
import type { ExecutionStatus as ExecutionStatusV1, WorkflowExecuteMode } from 'n8n-workflow';
/** A status added later reads as `unknown` rather than being guessed at. */
const V1_STATUS_BY_V2_STATUS = new Map<ExecutionStatus, ExecutionStatusV1>([
['queued', 'new'],
['running', 'running'],
['completed', 'success'],
['failed', 'error'],
['cancelled', 'canceled'],
]);
/** Anything not manual is a production run. */
const V1_MODE_BY_V2_MODE = new Map<ExecutionMode, WorkflowExecuteMode>([
['manual', 'manual'],
['production', 'trigger'],
]);
export function toV1Status(status: ExecutionStatus): ExecutionStatusV1 {
return V1_STATUS_BY_V2_STATUS.get(status) ?? 'unknown';
}
export function toV1Mode(mode: ExecutionMode): WorkflowExecuteMode {
return V1_MODE_BY_V2_MODE.get(mode) ?? 'trigger';
}
/**
* The v2 status codes that map onto one of `v1Statuses`, or `undefined` when no
* filter is given. An unfiltered list must not be narrowed to the mapped
* statuses, or a status the data plane adds later never shows up.
*/
export function resolveV2Statuses(v1Statuses?: ExecutionStatusV1[]): ExecutionStatus[] | undefined {
if (!v1Statuses?.length) return undefined;
return [...V1_STATUS_BY_V2_STATUS.entries()]
.filter(([, v1]) => v1Statuses.includes(v1))
.map(([v2]) => v2);
}
+23 -35
View File
@@ -3,7 +3,7 @@ import { z } from 'zod';
import { BadRequestError } from '@/errors/response-errors/bad-request.error';
import { isExecutionIdV2 } from './execution-id';
const timestamp = z.string().datetime({ offset: true });
/** An engine 1.0 execution ID: an auto-increment integer. */
const V1_ID = z
@@ -11,51 +11,39 @@ const V1_ID = z
.regex(/^[1-9]\d*$/)
.refine((id) => Number(id) <= 2147483647);
/** An engine 2.0 execution ID: a UUID. */
const V2_ID = z.string().refine(isExecutionIdV2);
/**
* A cursor for a list merging engine 1.0 (integer ID) and engine 2.0 (UUID) rows, since
* the two ID spaces don't sort against each other and must each track their own position.
*/
const Cursor = z
.object({
version: z.literal(1),
v1: z.object({ timestamp, id: V1_ID }).strict().optional(),
v2: z.object({ timestamp, id: z.string().uuid() }).strict().optional(),
})
.strict();
/** A position holds one ID space, because the two spaces do not sort against each other. */
const Cursor = z.union([
z.object({ version: z.literal(1), v1: V1_ID }).strict(),
z.object({ version: z.literal(1), v2: V2_ID }).strict(),
]);
export type ExecutionCursor = z.infer<typeof Cursor>;
export type ExecutionPosition = NonNullable<ExecutionCursor['v1']>;
/** Parse a cursor into the execution ID to page from, or `undefined` for the first page. */
export function parseExecutionCursor(value?: string): string | undefined {
if (value === undefined) return undefined;
let cursor: z.infer<typeof Cursor>;
/** Parse a cursor, or an empty one when the caller asks for the first page. */
export function parseExecutionCursor(value?: string): ExecutionCursor {
if (value === undefined) return { version: 1 };
try {
if (!value.length || value.length > 2048 || !/^[A-Za-z0-9_-]+$/.test(value)) {
throw new BadRequestError('Invalid execution cursor');
}
cursor = Cursor.parse(JSON.parse(Buffer.from(value, 'base64url').toString('utf8')));
return Cursor.parse(JSON.parse(Buffer.from(value, 'base64url').toString('utf8')));
} catch {
throw new BadRequestError('Invalid execution cursor');
}
// The executions table pages by its integer `id`, so a v2 position has nothing
// to page against yet. Nothing encodes one either - see `encodeCursorForId`.
if ('v2' in cursor) throw new BadRequestError('Unsupported execution cursor');
return cursor.v1;
}
/** Encode a cursor position. The ID shape picks the ID space it belongs to. */
export function encodeExecutionCursor(id: string): SerializedCursor {
const position = isExecutionIdV2(id) ? { v2: id } : { v1: id };
return Buffer.from(JSON.stringify({ version: 1, ...position })).toString(
'base64url',
) as SerializedCursor;
/** Whether a cursor marks a position, as opposed to asking for the first page. */
export function hasPosition(cursor?: ExecutionCursor): boolean {
return cursor?.v1 !== undefined || cursor?.v2 !== undefined;
}
/**
* Encode the cursor that continues a list from just after execution `id`, or `null` if `id`
* is an engine 2.0 execution ID — those aren't supported as a cursor position yet.
*/
export function encodeCursorForId(id: string): SerializedCursor | null {
if (isExecutionIdV2(id)) return null;
return encodeExecutionCursor(id);
export function encodeExecutionCursor(cursor: ExecutionCursor): SerializedCursor {
return Buffer.from(JSON.stringify(cursor)).toString('base64url') as SerializedCursor;
}
@@ -5,7 +5,7 @@ import { ExecutionRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import { ExecutionStatusList, type ExecutionStatus, type ExecutionSummary } from 'n8n-workflow';
import { encodeCursorForId } from './execution-cursor';
import { encodeExecutionCursor, type ExecutionCursor } from './execution-cursor';
export interface ListExecutionsResponse {
results: ExecutionSummary[];
@@ -21,7 +21,7 @@ export const COMPLETED_STATUSES = ExecutionStatusList.filter(
(status) => !CURRENT_STATUSES.includes(status),
);
/** Provides the logic for listing engine 1.0 executions. */
/** Provides the logic for listing engine v1 executions */
@Service()
export class ExecutionListV1Service {
constructor(
@@ -35,17 +35,24 @@ export class ExecutionListV1Service {
* Return also the total count of all executions that satisfy the query,
* and whether the total is an estimate or not.
*/
async findPageWithCount(query: ExecutionSummaries.RangeQuery): Promise<ListExecutionsResponse> {
const { range: _, ...countQuery } = query;
async findPageWithCount(
query: ExecutionSummaries.RangeQuery,
cursor?: ExecutionCursor,
): Promise<ListExecutionsResponse> {
const pagedQuery: ExecutionSummaries.RangeQuery = {
...query,
range: { ...query.range, beforeId: cursor?.v1?.id ?? query.range.beforeId },
};
const { range: _, ...countQuery } = pagedQuery;
const [results, executionCount] = await Promise.all([
this.executionRepository.findManyByRangeQuery(query),
this.executionRepository.findManyByRangeQuery(pagedQuery),
this.getExecutionsCountForQuery({ ...countQuery, kind: 'count' }),
]);
return {
results,
nextCursor: this.nextCursorFor(results, query.range.limit),
nextCursor: this.nextCursorFor(results, pagedQuery.range.limit),
...executionCount,
};
}
@@ -83,11 +90,13 @@ export class ExecutionListV1Service {
};
}
private async findCurrentExecutions(query: ExecutionSummaries.RangeQuery) {
async findCurrentExecutions(query: ExecutionSummaries.RangeQuery) {
const currentQuery: ExecutionSummaries.RangeQuery = {
...query,
// "current" is refetched in full on every page, so it ignores the cursor.
range: { limit: query.range.limit },
range: {
limit: query.range.limit,
},
status: CURRENT_STATUSES,
order: { top: 'running' }, // ensure limit cannot exclude running
};
@@ -100,7 +109,15 @@ export class ExecutionListV1Service {
if (rows.length < limit) return null;
const lastRow = rows[rows.length - 1];
return lastRow ? encodeCursorForId(lastRow.id) : null;
if (!lastRow) return null;
return encodeExecutionCursor({
version: 1,
v1: {
id: lastRow.id,
timestamp: new Date(lastRow.startedAt ?? lastRow.createdAt).toISOString(),
},
});
}
/**
@@ -0,0 +1,231 @@
import { compareExecutionListItems } from '@n8n/api-types';
import type { ExecutionSummaries } from '@n8n/db';
import { ExecutionListRepository, ExecutionRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import { hasGlobalScope } from '@n8n/permissions';
import type { ExecutionSummary } from 'n8n-workflow';
import assert from 'node:assert';
import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
import {
EngineV2ExecutionReader,
type EngineV2SearchQuery,
type V2Scope,
} from './engine-v2-execution-reader.service';
import type { ExecutionCursor, ExecutionPosition } from './execution-cursor';
import { isExecutionIdV2 } from './execution-id';
import {
CURRENT_STATUSES,
ExecutionListV1Service,
type ListExecutionsResponse,
} from './execution-list-v1.service';
import { mergeExecutionPages } from './merge-execution-pages';
class ExecutionListCtx {
constructor(
readonly query: Readonly<ExecutionSummaries.RangeQuery>,
readonly v2Reader: EngineV2ExecutionReader,
/** `null` keeps the data plane out of this request. */
readonly v2Scopes: V2Scope | null,
readonly cursor?: Readonly<ExecutionCursor>,
) {}
}
/** Provides the logic for listing executions from both engine v1 and v2 */
@Service()
export class ExecutionListV2Service {
constructor(
private readonly executionListV1Service: ExecutionListV1Service,
private readonly executionRepository: ExecutionRepository,
private readonly executionListRepository: ExecutionListRepository,
private readonly reader: EngineV2ExecutionReader,
private readonly dataPlane: EngineDataPlaneProxyService,
) {}
async findPageWithCount(
query: ExecutionSummaries.RangeQuery,
cursor?: ExecutionCursor,
): Promise<ListExecutionsResponse> {
const ctx = await this.buildCtx(query, cursor);
const result = await this.findPageMerged(ctx);
await this.resolveWorkflowNames(result.results);
return result;
}
/**
* Retrieves current (new & running) and completed (all other) executions
* from both v1 and v2 engines and merges the results.
*
* @param query the completed-only page query. The current block overrides the
* status, order and range itself, so that narrowing does not reach it.
*/
async findCurrentAndCompleted(
query: ExecutionSummaries.RangeQuery,
): Promise<ListExecutionsResponse> {
const ctx = await this.buildCtx(query);
const [currentMerged, completedMerged] = await Promise.all([
this.findCurrentMerged(ctx),
this.findPageMerged(ctx),
]);
const results = currentMerged.concat(completedMerged.results);
await this.resolveWorkflowNames(results);
return {
results,
nextCursor: completedMerged.nextCursor,
count: completedMerged.count,
estimated: completedMerged.estimated,
};
}
private async findCurrentMerged(ctx: ExecutionListCtx) {
const [v1Current, v2Current] = await Promise.all([
this.executionListV1Service.findCurrentExecutions(ctx.query),
this.findCurrentExecutionsV2(ctx),
]);
// Not capped to `limit`: each store contributes up to `limit` current rows,
// so the block can hold twice that. Current rows are few in practice.
return v1Current.concat(v2Current).sort((a, b) => {
const aRunning = a.status === 'running';
const bRunning = b.status === 'running';
if (aRunning !== bRunning) return aRunning ? -1 : 1;
return compareExecutionListItems(a, b);
});
}
private async findPageMerged(ctx: ExecutionListCtx): Promise<ListExecutionsResponse> {
const [v1Page, v2Page] = await Promise.all([this.v1Page(ctx), this.v2Page(ctx)]);
const merged = mergeExecutionPages([v1Page, v2Page], ctx.query.range.limit, ctx.cursor);
return {
results: merged.results,
nextCursor: merged.nextCursor,
// `-1` means the control plane could not count at all, so no total exists.
count: v1Page.count === -1 ? -1 : v1Page.count + v2Page.total,
estimated: v1Page.estimated,
};
}
/**
* Which workflows the data plane may be searched for, or `null` to keep it out
* of this request.
*
* The data plane search has no equivalent for a metadata, annotation, vote, or
* workflow-version filter, so a query using one is answered from the control
* plane alone. That silently drops every engine 2.0 execution from the result.
*/
async resolveV2Scope(query: ExecutionSummaries.RangeQuery): Promise<V2Scope | null> {
if (
query.metadata?.length ||
query.annotationTags?.length ||
query.vote ||
query.workflowVersionId !== undefined
)
return null;
const user = query.user;
assert(user);
if (
hasGlobalScope(user, 'workflow:read') &&
!query.workflowId &&
!query.projectId &&
query.isArchived === undefined &&
!query.workflowBooleanSettings?.length
)
return 'all';
return await this.executionListRepository.findWorkflowIdsForExecutionList(query);
}
private async v1Page(ctx: ExecutionListCtx) {
const { range: _, ...countQuery } = ctx.query;
const [rows, executionCount] = await Promise.all([
this.executionRepository.findManyByRangeQuery({
...ctx.query,
// One row past the page tells the merge whether the store has more.
range: {
limit: ctx.query.range.limit + 1,
beforeId: ctx.cursor?.v1?.id ?? ctx.query.range.beforeId,
},
}),
this.executionListV1Service.getExecutionsCountForQuery({ ...countQuery, kind: 'count' }),
]);
return {
items: rows.slice(0, ctx.query.range.limit),
hasMore: rows.length > ctx.query.range.limit,
count: executionCount.count,
estimated: executionCount.estimated,
};
}
private async v2Page(ctx: ExecutionListCtx) {
if (!ctx.v2Scopes) return { items: [], total: 0, hasMore: false };
// The page reports a total, so `count` can add both stores together.
return await ctx.v2Reader.findMany(this.toV2Query(ctx.query, ctx.cursor?.v2), ctx.v2Scopes, {
includeTotal: true,
});
}
/** Narrows a control-plane range query down to what the data plane search understands. */
private toV2Query(
query: ExecutionSummaries.RangeQuery,
before?: ExecutionPosition,
): EngineV2SearchQuery {
return {
status: query.status,
mode: query.mode,
startedAfter: query.startedAfter,
startedBefore: query.startedBefore,
before,
limit: query.range.limit,
};
}
/** Resolves and assigns `workflowName` on every v2 result, in place. */
private async resolveWorkflowNames(results: ExecutionSummary[]): Promise<void> {
const v2Results = results.filter((item) => isExecutionIdV2(item.id));
if (!v2Results.length) return;
const uniqV2WorkflowIds = [...new Set(v2Results.map((item) => item.workflowId))];
const workflowNames =
await this.executionListRepository.findNamesForExecutionList(uniqV2WorkflowIds);
const namesById = new Map(workflowNames.map((workflow) => [workflow.id, workflow.name]));
for (const item of v2Results) item.workflowName = namesById.get(item.workflowId);
}
private async findCurrentExecutionsV2(ctx: ExecutionListCtx) {
if (!ctx.v2Scopes) return [];
// No total: the current block is never paged, and `count` excludes it.
const currentExecutions = await ctx.v2Reader.findMany(
{
...this.toV2Query(ctx.query, ctx.cursor?.v2),
status: CURRENT_STATUSES,
order: { top: 'running' }, // ensure limit cannot exclude running
},
ctx.v2Scopes,
);
return currentExecutions.items;
}
private async buildCtx(query: ExecutionSummaries.RangeQuery, cursor?: ExecutionCursor) {
// The router only picks this provider while the data plane is up.
assert(this.dataPlane.isAvailable());
return new ExecutionListCtx(query, this.reader, await this.resolveV2Scope(query), cursor);
}
}
@@ -1,48 +1,72 @@
import type { ExecutionSummaries, User } from '@n8n/db';
import { Service } from '@n8n/di';
import { Container, Service } from '@n8n/di';
import type { Scope } from '@n8n/permissions';
import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.service';
import { RoleService } from '@/services/role.service';
import { WorkflowSharingService } from '@/workflows/workflow-sharing.service';
import { hasPosition, type ExecutionCursor } from './execution-cursor';
import {
COMPLETED_STATUSES,
ExecutionListV1Service,
type ListExecutionsResponse,
} from './execution-list-v1.service';
import type { ExecutionListV2Service } from './execution-list-v2.service';
/**
* Entry point for execution list requests. Combine v1+v2 executions.
* Entry point for execution list requests. Routes to the v1 provider, or to
* the v1+v2 provider that combines both.
*
* Engine v1 executions live in the CP, engine v2 executions live in the DP.
* This service reads from both stores and combines the results.
*/
@Service()
export class ExecutionListService {
constructor(
private readonly v1Service: ExecutionListV1Service,
private readonly dataPlane: EngineDataPlaneProxyService,
private readonly workflowSharingService: WorkflowSharingService,
private readonly roleService: RoleService,
) {}
/** Lists executions for the executions page. */
async listExecutionsForUI(query: ExecutionSummaries.RangeQuery): Promise<ListExecutionsResponse> {
// Without a status filter, the first page combines two blocks: current on
// top, then a completed-only page whose count excludes the current rows.
// A cursor page is not split, so it spans every status.
const combinesCurrentAndCompleted = !query.status?.length && !query.range.beforeId;
async listExecutionsForUI(
query: ExecutionSummaries.RangeQuery,
cursor?: ExecutionCursor,
): Promise<ListExecutionsResponse> {
// Without a status filter the list combines two blocks: current on top, then
// a completed-only page. Every page keeps that completed filter, so "load
// more" never repeats the current rows and the count stays completed-only.
// Only the first page prepends the current block.
const combinesCurrentAndCompleted = !query.status?.length;
const hasPositionInCursor = hasPosition(cursor);
return combinesCurrentAndCompleted
? await this.v1Service.findCurrentAndCompleted({ ...query, status: COMPLETED_STATUSES })
: await this.v1Service.findPageWithCount(query);
// The `startedAt` order is what both stores page against, so the merge and
// the cursor agree on what "next" means.
const pageQuery: ExecutionSummaries.RangeQuery = {
...query,
order: { startedAt: 'DESC' },
...(combinesCurrentAndCompleted ? { status: COMPLETED_STATUSES } : {}),
};
const provider = await this.provider();
return combinesCurrentAndCompleted && !hasPositionInCursor
? await provider.findCurrentAndCompleted(pageQuery)
: await provider.findPageWithCount(pageQuery, cursor);
}
/**
* Lists a single "page" of executions and full count. A page is `limit`
* executions starting from the cursor position.
* executions starting from `cursor` position.
*/
async findPageWithCount(query: ExecutionSummaries.RangeQuery): Promise<ListExecutionsResponse> {
return await this.v1Service.findPageWithCount(query);
async findPageWithCount(
query: ExecutionSummaries.RangeQuery,
cursor?: ExecutionCursor,
): Promise<ListExecutionsResponse> {
const provider = await this.provider();
return await provider.findPageWithCount(query, cursor);
}
/** Assigns the caller's scopes for each listed execution's workflow, in place. */
@@ -70,4 +94,13 @@ export class ExecutionListService {
const workflowRoles = await this.roleService.rolesWithScope('workflow', [scope]);
return { scopes: [scope], projectRoles, workflowRoles };
}
/** The provider for this request: v1 alone, or v1 merged with v2. */
private async provider(): Promise<ExecutionListV1Service | ExecutionListV2Service> {
if (!this.dataPlane.isAvailable()) return this.v1Service;
const { ExecutionListV2Service } = await import('./execution-list-v2.service.js');
return Container.get(ExecutionListV2Service);
}
}
@@ -1,15 +1,15 @@
import type { ExecutionListPaginationQuery } from '@n8n/api-types';
import type { SerializedCursor } from '@n8n/api-types';
import type { AuthenticatedRequest, ExecutionSummaries, ExecutionEntity } from '@n8n/db';
import type { AnnotationVote, ExecutionStatus, WorkflowExecuteMode } from 'n8n-workflow';
import type { ExecutionCursor } from '@/executions/execution-cursor';
export declare namespace ExecutionRequest {
namespace QueryParams {
type GetMany = {
cursor?: ExecutionListPaginationQuery['cursor'];
cursor?: SerializedCursor;
filter: string; // stringified `FilterFields`
limit: string;
lastId: string;
firstId: string;
};
}
@@ -32,6 +32,7 @@ export declare namespace ExecutionRequest {
type GetMany = AuthenticatedRequest<{}, {}, {}, QueryParams.GetMany> & {
rangeQuery: ExecutionSummaries.RangeQuery; // parsed from query params
cursor?: ExecutionCursor;
};
type GetOne = AuthenticatedRequest<RouteParams.ExecutionId>;
@@ -23,10 +23,10 @@ import { validateExecutionUpdatePayload } from './validation';
export class ExecutionsController {
constructor(
private readonly executionService: ExecutionService,
private readonly executionListService: ExecutionListService,
private readonly enterpriseExecutionService: EnterpriseExecutionsService,
private readonly workflowSharingService: WorkflowSharingService,
private readonly license: License,
private readonly executionListService: ExecutionListService,
) {}
private async getAccessibleWorkflowIds(user: User, scope: Scope) {
@@ -35,7 +35,7 @@ export class ExecutionsController {
@Get('/', { middlewares: [parseRangeQuery] })
async getMany(req: ExecutionRequest.GetMany) {
const { rangeQuery: query } = req;
const { rangeQuery: query, cursor } = req;
query.user = req.user;
query.sharingOptions = await this.executionListService.buildSharingOptions('workflow:read');
@@ -46,7 +46,7 @@ export class ExecutionsController {
}
const [executions, concurrentExecutionsCount] = await Promise.all([
this.executionListService.listExecutionsForUI(query),
this.executionListService.listExecutionsForUI(query, cursor),
this.executionService.getConcurrentExecutionsCount(),
]);
await this.executionListService.addScopes(
@@ -0,0 +1,72 @@
import { compareExecutionListItems } from '@n8n/api-types';
import type { ExecutionSummary } from 'n8n-workflow';
import { encodeExecutionCursor, type ExecutionCursor } from './execution-cursor';
import { isExecutionIdV2 } from './execution-id';
export interface ExecutionPage {
items: ExecutionSummary[];
hasMore: boolean;
}
function positionOf(item: ExecutionSummary) {
return {
id: item.id,
timestamp: new Date(item.startedAt ?? item.createdAt).toISOString(),
};
}
/**
* Merges one page from each store into a single newest-first page.
*
* Each store is asked for `limit` rows, so up to `limit` candidates are fetched
* and never used. The cursor therefore carries a separate position per store,
* each marking that store's last **included** row — not its last fetched one.
* A candidate that missed the cut is simply fetched again on the next page.
*
* ```text
* fetched (limit 3) merged, newest first
* ───────────────────────── ────────────────────────────────
* v1 10:05 10:03 10:01 10:05 v1 ┐
* v2 10:04 10:02 10:00 10:04 v2 │ page 1 ← last v2 included
* 10:03 v1 ┘ ← last v1 included
* ╌╌╌╌╌╌ cut ╌╌╌╌╌╌
* 10:02 v2 dropped
* 10:01 v1 dropped
* 10:00 v2 dropped
*
* nextCursor = { v1: 10:03, v2: 10:04 }
*
* Note v2's mark is 10:04, not the 10:00 it read up to, so page 2 re-reads
* 10:02 and 10:00. Each store resumes just after its own mark:
*
* v1 < 10:03 → 10:01 ... 10:02 v2 ┐
* v2 < 10:04 → 10:02 10:00 ... 10:01 v1 │ page 2
* 10:00 v2 ┘
* ```
*
* A page drawn from one store only leaves the other store's position untouched,
* so that store resumes where the earlier page left it.
*/
export function mergeExecutionPages(
pages: ExecutionPage[],
limit: number,
cursor?: ExecutionCursor,
) {
const v1v2MergedAndSorted = pages.flatMap((page) => page.items).sort(compareExecutionListItems);
const results = v1v2MergedAndSorted.slice(0, limit);
const resultsReversed = Array.from(results).reverse();
const lastIncludedV1 = resultsReversed.find((item) => !isExecutionIdV2(item.id));
const lastIncludedV2 = resultsReversed.find((item) => isExecutionIdV2(item.id));
const nextCursor: ExecutionCursor = cursor ? { ...cursor } : { version: 1 };
if (lastIncludedV1) nextCursor.v1 = positionOf(lastIncludedV1);
if (lastIncludedV2) nextCursor.v2 = positionOf(lastIncludedV2);
const hasMore = v1v2MergedAndSorted.length > limit || pages.some((page) => page.hasMore);
return {
results,
nextCursor: hasMore && results.length ? encodeExecutionCursor(nextCursor) : null,
};
}
@@ -47,7 +47,8 @@ export const parseRangeQuery = (req: Request, res: Response, next: NextFunction)
if (req.query.cursor !== undefined && typeof req.query.cursor !== 'string')
throw new BadRequestError('Invalid execution cursor');
const beforeId = parseExecutionCursor(req.query.cursor);
const cursor =
req.query.cursor === undefined ? undefined : parseExecutionCursor(req.query.cursor);
const pageLimit = limit === undefined ? 20 : Number(limit);
if (!Number.isInteger(pageLimit) || pageLimit < 1 || pageLimit > 100)
@@ -57,11 +58,10 @@ export const parseRangeQuery = (req: Request, res: Response, next: NextFunction)
kind: 'range',
range: {
limit: pageLimit,
...(beforeId ? { beforeId } : {}),
},
});
Object.assign(req, { rangeQuery });
Object.assign(req, { rangeQuery, cursor });
next();
} catch (error) {
if (error instanceof Error) {
@@ -62,6 +62,28 @@ describe('EngineDataPlaneClient', () => {
client = newClient();
});
it('posts search filters without following redirects', async () => {
const body = { workflowIds: ['wf'], mode: 'webhook', includeTotal: true, limit: 20 };
const result = { items: [], hasMore: false, total: 0 };
respondWith(200, result);
await expect(client.searchExecutions(body)).resolves.toEqual(result);
expect(http.request).toHaveBeenCalledWith(
expect.objectContaining({
url: '/api/workflow-executions/search',
method: 'POST',
body,
disableFollowRedirect: true,
}),
);
});
it('propagates search failures', async () => {
respondWith(503, {});
await expect(client.searchExecutions({ workflowIds: 'all', limit: 20 })).rejects.toThrow(
OperationalError,
);
});
describe('startExecution', () => {
it('posts the request to the engine and returns the execution id', async () => {
respondWith(201, { executionId: 'exec-1' });
@@ -9,6 +9,8 @@ import type {
ExecutionSnapshot,
StartExecutionRequest,
StartExecutionResult,
SearchExecutionsRequest,
SearchExecutionsResponse,
} from '@n8n/engine';
import { mintIdentityToken } from '@n8n/engine';
import { InstanceSettings } from 'n8n-core';
@@ -105,6 +107,21 @@ export class EngineDataPlaneClient implements EngineDataPlaneProvider {
return response.body as ExecutionSnapshot;
}
/** Lists executions stored in the engine's data plane, matching `request`. */
async searchExecutions(request: SearchExecutionsRequest): Promise<SearchExecutionsResponse> {
const response = await this.http.request<SearchExecutionsResponse | EngineErrorResponse>({
url: '/api/workflow-executions/search',
method: 'POST',
body: request,
json: true,
returnFullResponse: true,
ignoreHttpStatusErrors: true,
disableFollowRedirect: true,
});
if (response.statusCode >= 300) throw this.toError(response.statusCode, response.body);
return response.body as SearchExecutionsResponse;
}
private toError(statusCode: number, body: unknown): Error {
const { error, reason } = this.parseErrorResponse(body);
const detail = reason ?? error;
@@ -26,23 +26,25 @@ const createExecution = (overrides: Partial<ExecutionSummary> = {}): ExecutionSu
describe('search-executions MCP tool', () => {
const user = Object.assign(new User(), { id: 'user-1' });
const uuid = '01992380-0000-7000-8000-000000000001';
const time = '2026-09-07T12:00:00.000Z';
let executionListService: ExecutionListService;
let workflowFinderService: WorkflowFinderService;
let telemetry: Telemetry;
beforeEach(() => {
executionListService = mockInstance(ExecutionListService, {
buildSharingOptions: vi.fn().mockResolvedValue({
scopes: ['workflow:read'],
projectRoles: ['project:editor'],
workflowRoles: ['workflow:editor'],
}),
findPageWithCount: vi.fn().mockResolvedValue({
results: [],
count: 0,
estimated: false,
nextCursor: null,
}),
buildSharingOptions: vi.fn().mockResolvedValue({
scopes: ['workflow:read'],
projectRoles: ['project:editor'],
workflowRoles: ['workflow:editor'],
}),
});
workflowFinderService = mockInstance(WorkflowFinderService, {
findWorkflowForUser: vi.fn().mockResolvedValue({
@@ -164,13 +166,23 @@ describe('search-executions MCP tool', () => {
expect(query.range.limit).toBe(200);
});
test('pages from the execution ID the cursor encodes', async () => {
const cursor = encodeExecutionCursor('50');
// The tool does not bound the page itself. It hands the decoded cursor over,
// and the list service applies each store's position to that store.
test('forwards the decoded cursor instead of bounding the range', async () => {
const position = {
version: 1 as const,
v1: { id: '50', timestamp: time },
v2: { id: uuid, timestamp: time },
};
await createTool().handler({ cursor } as never, {} as never);
await createTool().handler({ cursor: encodeExecutionCursor(position) } as never, {} as never);
expect(executionListService.findPageWithCount).toHaveBeenCalledWith(
expect.anything(),
position,
);
const query = (executionListService.findPageWithCount as Mock).mock.calls[0][0];
expect(query.range.beforeId).toBe('50');
expect(query.range).toEqual({ limit: 200 });
});
test('returns an error for an invalid cursor', async () => {
@@ -191,7 +203,7 @@ describe('search-executions MCP tool', () => {
});
});
test('delegates sharing options to executionListService.buildSharingOptions', async () => {
test('delegates sharing options to the list service', async () => {
await createTool().handler({} as never, {} as never);
expect(executionListService.buildSharingOptions).toHaveBeenCalledWith('workflow:read');
@@ -486,6 +486,9 @@ export class McpService {
);
registerIfAllowed(getExecutionTool);
// TODO(CAT-4510): the search lists engine 2.0 executions, but
// `get_workflow_execution` above still reads only the control plane, so an
// agent cannot fetch a v2 result it just found.
const searchExecutionsTool = createSearchExecutionsTool(
user,
this.executionListService,
@@ -118,16 +118,14 @@ export const createSearchExecutionsTool = (
const safeLimit = Math.min(Math.max(1, limit), MAX_RESULTS);
const sharingOptions = await executionListService.buildSharingOptions('workflow:read');
const beforeId = parseExecutionCursor(cursor);
const parsedCursor = parseExecutionCursor(cursor);
const query = {
kind: 'range' as const,
user,
sharingOptions,
range: {
limit: safeLimit,
...(beforeId ? { beforeId } : {}),
},
// The cursor goes to the list service, which bounds each store itself.
range: { limit: safeLimit },
...(workflowId ? { workflowId } : {}),
...(status?.length ? { status } : {}),
...(startedAfter ? { startedAfter } : {}),
@@ -137,7 +135,7 @@ export const createSearchExecutionsTool = (
};
const { results, count, estimated, nextCursor } =
await executionListService.findPageWithCount(query);
await executionListService.findPageWithCount(query, parsedCursor);
const data = results.map((execution) => ({
id: execution.id,
@@ -51,6 +51,24 @@ describe('EngineDataPlaneProxyService', () => {
await expect(proxy.getExecution(executionId)).resolves.toBeUndefined();
});
it('returns an empty search without a provider', async () => {
await expect(proxy.searchExecutions({ workflowIds: 'all', limit: 20 })).resolves.toEqual({
items: [],
nextCursor: null,
total: 0,
});
});
it('forwards search filters and propagates provider failures', async () => {
const provider = mock<EngineDataPlaneProvider>();
proxy.registerProvider(provider);
provider.searchExecutions.mockRejectedValue(new Error('unavailable'));
await expect(proxy.searchExecutions({ workflowIds: ['wf'], limit: 20 })).rejects.toThrow(
'unavailable',
);
expect(provider.searchExecutions).toHaveBeenCalledWith({ workflowIds: ['wf'], limit: 20 });
});
it('delegates a read to the registered provider', async () => {
const provider = mock<EngineDataPlaneProvider>();
const snapshot = mock<ExecutionSnapshot>({ id: executionId });
@@ -1,5 +1,11 @@
import { Service } from '@n8n/di';
import type { ExecutionSnapshot, StartExecutionRequest, StartExecutionResult } from '@n8n/engine';
import type {
ExecutionSnapshot,
StartExecutionRequest,
StartExecutionResult,
SearchExecutionsRequest,
SearchExecutionsResponse,
} from '@n8n/engine';
import { UserError } from 'n8n-workflow';
import type { ExecutionIdV2 } from '@/executions/execution-id';
@@ -11,6 +17,7 @@ import type { ExecutionIdV2 } from '@/executions/execution-id';
* runs in the same process, so this stays a network-shaped contract.
*/
export interface EngineDataPlaneProvider {
searchExecutions(request: SearchExecutionsRequest): Promise<SearchExecutionsResponse>;
startExecution(request: StartExecutionRequest): Promise<StartExecutionResult>;
/**
@@ -44,6 +51,11 @@ export class EngineDataPlaneProxyService implements EngineDataPlaneProvider {
return this.provider !== null;
}
async searchExecutions(request: SearchExecutionsRequest): Promise<SearchExecutionsResponse> {
if (!this.provider) return { items: [], nextCursor: null, total: 0 };
return await this.provider.searchExecutions(request);
}
async startExecution(request: StartExecutionRequest): Promise<StartExecutionResult> {
if (!this.provider) {
throw new UserError(
@@ -33,6 +33,7 @@ const SET_NAME = 'Edit Fields';
const startExecution = vi.fn();
const getExecution = vi.fn();
const searchExecutions = vi.fn();
let builder: User;
@@ -87,7 +88,11 @@ const handOff = async (workflow: WorkflowEntity, items: INodeExecutionData[][])
beforeAll(async () => {
await initNodeTypes();
Container.get(EngineDataPlaneProxyService).registerProvider({ startExecution, getExecution });
Container.get(EngineDataPlaneProxyService).registerProvider({
startExecution,
getExecution,
searchExecutions,
});
});
beforeEach(async () => {
@@ -68,7 +68,11 @@ beforeAll(async () => {
webhookTestEndpoint = Container.get(GlobalConfig).endpoints.webhookTest;
await Container.get(CacheService).init();
Container.get(EngineDataPlaneProxyService).registerProvider({ startExecution, getExecution });
Container.get(EngineDataPlaneProxyService).registerProvider({
startExecution,
getExecution,
searchExecutions: vi.fn().mockResolvedValue({ items: [], nextCursor: null, total: 0 }),
});
// `/webhook-test/*` is mounted only when a server opts into test webhooks.
class EditorFacingWebhookServer extends WebhookServer {
@@ -16,11 +16,12 @@ import { ExecutionService } from '@/executions/execution.service';
import { annotateExecution, createAnnotationTags, createExecution } from './shared/db/executions';
import { createMember, createOwner } from './shared/db/users';
/** A status-less query, narrowed the way the list service narrows it when the two blocks combine. */
/** A status-less query, narrowed the way the list router narrows it when the two blocks combine. */
const completedPageQuery = (
query: ExecutionSummaries.RangeQuery,
): ExecutionSummaries.RangeQuery => ({
...query,
order: { startedAt: 'DESC' },
status: COMPLETED_STATUSES,
});
@@ -19,6 +19,7 @@ import { EngineDataPlaneProxyService } from '@/services/engine-data-plane-proxy.
import { WaitTracker } from '@/wait-tracker';
import {
createExecution,
createSuccessfulExecution,
createWaitingExecution,
getAllExecutions,
@@ -96,6 +97,86 @@ describe('GET /executions', () => {
const response = await testServer.authAgentFor(member).get('/executions').expect(200);
expect(response.body.data.results[0].scopes).toContain('workflow:execute');
});
describe('paging without a status filter', () => {
/** 2 running plus `completed` successful executions, newest id last. */
const seed = async (completed: number) => {
const workflow = await createWorkflow({}, owner);
await createExecution({ status: 'running', stoppedAt: undefined }, workflow);
await createExecution({ status: 'running', stoppedAt: undefined }, workflow);
for (let i = 0; i < completed; i++) {
await createExecution({ status: 'success' }, workflow);
}
};
test('reports the current block once and counts only completed rows', async () => {
await seed(5);
const response = await testServer
.authAgentFor(owner)
.get('/executions')
.query({ limit: 2 })
.expect(200);
const { results, count, nextCursor } = response.body.data;
expect(results.filter((r: { status: string }) => r.status === 'running')).toHaveLength(2);
// The count excludes the current block, so paging is over completed rows only.
expect(count).toBe(5);
expect(nextCursor).not.toBeNull();
});
test('keeps running executions out of later pages', async () => {
await seed(5);
const first = await testServer
.authAgentFor(owner)
.get('/executions')
.query({ limit: 2 })
.expect(200);
const second = await testServer
.authAgentFor(owner)
.get('/executions')
.query({ limit: 2, cursor: first.body.data.nextCursor })
.expect(200);
expect(second.body.data.results).toHaveLength(2);
expect(second.body.data.results.map((r: { status: string }) => r.status)).toEqual([
'success',
'success',
]);
expect(second.body.data.count).toBe(5);
});
test('walks every completed row exactly once', async () => {
await seed(5);
const seen: string[] = [];
let cursor: string | null = null;
// 5 rows at 2 per page needs 3 requests. A cursor that fails to advance
// would page forever, so cap the walk and assert on the cap.
let requests = 0;
do {
const response = await testServer
.authAgentFor(owner)
.get('/executions')
.query({ limit: 2, ...(cursor ? { cursor } : {}) })
.expect(200);
const data = response.body.data as {
results: Array<{ id: string; status: string }>;
nextCursor: string | null;
};
seen.push(...data.results.filter((r) => r.status === 'success').map((r) => r.id));
cursor = data.nextCursor;
} while (cursor && ++requests < 5);
expect(requests).toBeLessThan(4);
expect(seen).toHaveLength(5);
expect(new Set(seen).size).toBe(5);
});
});
});
describe('GET /executions/:id', () => {
@@ -148,7 +229,11 @@ describe('GET /executions/:id', () => {
const getExecution = vi.fn();
beforeAll(() => {
Container.get(EngineDataPlaneProxyService).registerProvider({ startExecution, getExecution });
Container.get(EngineDataPlaneProxyService).registerProvider({
startExecution,
getExecution,
searchExecutions: vi.fn().mockResolvedValue({ items: [], nextCursor: null, total: 0 }),
});
});
beforeEach(() => {
@@ -0,0 +1,247 @@
import {
createTeamProject,
createWorkflow,
linkUserToProject,
mockInstance,
testDb,
} from '@n8n/backend-test-utils';
import type { ExecutionSummaries, User } from '@n8n/db';
import { ExecutionListRepository, ExecutionRepository, SharedWorkflowRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import type { ExecutionListItem } from '@n8n/engine';
import { mock } from 'vitest-mock-extended';
import { ConcurrencyControlService } from '@/concurrency/concurrency-control.service';
import { ExecutionListService } from '@/executions/execution-list.service';
import {
EngineDataPlaneProxyService,
type EngineDataPlaneProvider,
} from '@/services/engine-data-plane-proxy.service';
import { WaitTracker } from '@/wait-tracker';
import { createExecution } from '../shared/db/executions';
import { createMember, createOwner } from '../shared/db/users';
import { setupTestServer } from '../shared/utils';
mockInstance(WaitTracker);
mockInstance(ConcurrencyControlService);
const server = setupTestServer({ endpointGroups: ['executions'] });
const provider = mock<EngineDataPlaneProvider>();
let owner: User;
let member: User;
const time = '2026-09-07T12:00:00.000Z';
const uuid = '01992380-0000-7000-8000-000000000001';
beforeEach(async () => {
await testDb.truncate(['ExecutionEntity', 'WorkflowEntity', 'SharedWorkflow']);
owner = await createOwner();
member = await createMember();
server.license.reset();
provider.searchExecutions.mockReset();
provider.searchExecutions.mockResolvedValue({ items: [], nextCursor: null, total: 0 });
Container.get(EngineDataPlaneProxyService).registerProvider(provider);
});
const row = (workflowId: string): ExecutionListItem => ({
id: uuid,
workflowId,
status: 'completed',
mode: 'manual',
createdAt: time,
updatedAt: time,
finishedAt: time,
});
describe('editor execution visibility', () => {
it('applies a project move on the next request', async () => {
const home = await createTeamProject();
const destination = await createTeamProject();
await linkUserToProject(member, home, 'project:admin');
const workflow = await createWorkflow({}, home);
const filter = JSON.stringify({ status: ['success'], workflowId: workflow.id });
await server.authAgentFor(member).get('/executions').query({ filter }).expect(200);
expect(provider.searchExecutions).toHaveBeenCalledWith(
expect.objectContaining({ workflowIds: [workflow.id] }),
);
await Container.get(SharedWorkflowRepository).update(
{ workflowId: workflow.id, projectId: home.id },
{ projectId: destination.id },
);
provider.searchExecutions.mockClear();
await server.authAgentFor(member).get('/executions').query({ filter }).expect(200);
expect(provider.searchExecutions).not.toHaveBeenCalled();
});
it('uses all for the owner and omits the DP for a member without access', async () => {
const workflow = await createWorkflow({}, owner);
provider.searchExecutions.mockResolvedValue({
items: [row(workflow.id)],
nextCursor: null,
total: 1,
});
await server
.authAgentFor(owner)
.get('/executions')
.query({ filter: JSON.stringify({ status: ['success'] }) })
.expect(200);
expect(provider.searchExecutions).toHaveBeenLastCalledWith(
expect.objectContaining({ workflowIds: 'all' }),
);
provider.searchExecutions.mockClear();
await server
.authAgentFor(member)
.get('/executions')
.query({ filter: JSON.stringify({ status: ['success'], workflowId: workflow.id }) })
.expect(200);
expect(provider.searchExecutions).not.toHaveBeenCalled();
});
it('includes a workflow shared into the selected project and removes access on the next request', async () => {
const home = await createTeamProject();
const visible = await createTeamProject();
await linkUserToProject(member, visible, 'project:admin');
const workflow = await createWorkflow({}, home);
await Container.get(SharedWorkflowRepository).save({
workflowId: workflow.id,
projectId: visible.id,
role: 'workflow:editor',
});
provider.searchExecutions.mockResolvedValue({
items: [row(workflow.id)],
total: 1,
nextCursor: null,
});
const filter = JSON.stringify({ status: ['success'], projectId: visible.id });
const response = await server
.authAgentFor(member)
.get('/executions')
.query({ filter })
.expect(200);
const body = response.body as {
data: { results: Array<{ id: string; workflowName: string }>; count: number };
};
expect(body.data).toMatchObject({
results: [{ id: uuid, workflowName: workflow.name }],
count: 1,
});
expect(provider.searchExecutions).toHaveBeenLastCalledWith(
expect.objectContaining({ workflowIds: [workflow.id] }),
);
await Container.get(SharedWorkflowRepository).delete({
workflowId: workflow.id,
projectId: visible.id,
});
provider.searchExecutions.mockClear();
await server.authAgentFor(member).get('/executions').query({ filter }).expect(200);
expect(provider.searchExecutions).not.toHaveBeenCalled();
});
it('resolves archive and workflow setting filters in the CP', async () => {
const workflow = await createWorkflow({ settings: { availableInMCP: true } }, owner);
const sharingOptions =
await Container.get(ExecutionListService).buildSharingOptions('workflow:read');
const query: ExecutionSummaries.RangeQuery = {
kind: 'range',
range: { limit: 2 },
user: owner,
sharingOptions,
workflowId: workflow.id,
isArchived: false,
workflowBooleanSettings: [{ key: 'availableInMCP', value: true }],
};
const workflows = Container.get(ExecutionListRepository);
expect(await workflows.findWorkflowIdsForExecutionList(query)).toEqual([workflow.id]);
expect(await workflows.findWorkflowIdsForExecutionList({ ...query, isArchived: true })).toEqual(
[],
);
expect(
await workflows.findWorkflowIdsForExecutionList({
...query,
workflowBooleanSettings: [{ key: 'availableInMCP', value: false }],
}),
).toEqual([]);
});
});
describe('editor v1 source cursor', () => {
it('walks mixed REST pages with equal timestamps without gaps or duplicates', async () => {
const workflow = await createWorkflow({}, owner);
const v1Ids: string[] = [];
for (let i = 0; i < 5; i++) {
const execution = await createExecution(
{ createdAt: new Date(time), startedAt: new Date(time), status: 'success' },
workflow,
);
v1Ids.push(execution.id);
}
const v2Rows = ['3', '2', '1'].map((suffix) => ({
...row(workflow.id),
id: `${uuid.slice(0, -1)}${suffix}`,
}));
provider.searchExecutions.mockImplementation(async (query) => {
const remaining = v2Rows.filter(
(item) =>
!query.before ||
item.createdAt < query.before.createdAt ||
(item.createdAt === query.before.createdAt && item.id < query.before.id),
);
const limit = query.limit ?? 20;
const page = remaining.slice(0, limit);
const last = page.at(-1);
return {
items: page,
nextCursor:
remaining.length > limit && last ? { createdAt: last.createdAt, id: last.id } : null,
total: v2Rows.length,
};
});
const seen: string[] = [];
let cursor: string | null = null;
for (let i = 0; i < 4; i++) {
const response = await server
.authAgentFor(owner)
.get('/executions')
.query({
filter: JSON.stringify({ workflowId: workflow.id, status: ['success'] }),
limit: 2,
...(cursor ? { cursor } : {}),
})
.expect(200);
const { data } = response.body as {
data: { results: Array<{ id: string }>; count: number; nextCursor: string | null };
};
expect(data.count).toBe(8);
seen.push(...data.results.map((item) => item.id));
cursor = data.nextCursor;
if (i < 3) expect(cursor).not.toBeNull();
}
expect(cursor).toBeNull();
expect(seen).toEqual([...v2Rows.map((item) => item.id), ...v1Ids.reverse()]);
});
it('walks equal timestamps and a null start time in numeric ID order', async () => {
const workflow = await createWorkflow({}, owner);
const ids: string[] = [];
for (let i = 0; i < 5; i++) {
const execution = await createExecution(
{ createdAt: new Date(time), startedAt: i % 2 ? null : new Date(time), status: 'success' },
workflow,
);
ids.push(execution.id);
}
const repository = Container.get(ExecutionRepository);
const query: ExecutionSummaries.RangeQuery = {
kind: 'range',
user: owner,
workflowId: workflow.id,
range: { limit: 2 },
order: { startedAt: 'DESC' },
};
const seen: string[] = [];
for (let i = 0; i < 3; i++) {
const items = await repository.findManyByRangeQuery(query);
seen.push(...items.map((item) => item.id));
query.range.beforeId = items.at(-1)!.id;
}
expect(seen).toEqual(ids.reverse());
expect(await repository.fetchCount({ ...query, kind: 'count' })).toBe(5);
});
});
@@ -5027,7 +5027,11 @@ describe('POST /workflows/:workflowId/run', () => {
const getExecution = vi.fn();
beforeAll(() => {
Container.get(EngineDataPlaneProxyService).registerProvider({ startExecution, getExecution });
Container.get(EngineDataPlaneProxyService).registerProvider({
startExecution,
getExecution,
searchExecutions: vi.fn().mockResolvedValue({ items: [], nextCursor: null, total: 0 }),
});
});
beforeEach(() => {
+1 -1
View File
@@ -4255,7 +4255,7 @@ export interface ExecutionSummary {
};
usedPrivateCredentials?: boolean;
annotation?: {
vote: AnnotationVote;
vote?: AnnotationVote | null;
tags: Array<{
id: string;
name: string;