diff --git a/packages/@n8n/db/src/repositories/execution-list.repository.ts b/packages/@n8n/db/src/repositories/execution-list.repository.ts new file mode 100644 index 00000000000..c4dc8d3c34d --- /dev/null +++ b/packages/@n8n/db/src/repositories/execution-list.repository.ts @@ -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 { + 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 { + 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, + user: User, + sharingOptions: NonNullable, + ): 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, + 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>> { + if (ids.length === 0) return []; + + return await this.find({ select: ['id', 'name'], where: { id: In(ids) } }); + } +} diff --git a/packages/@n8n/db/src/repositories/index.ts b/packages/@n8n/db/src/repositories/index.ts index 673c5cb4001..d5e624d9113 100644 --- a/packages/@n8n/db/src/repositories/index.ts +++ b/packages/@n8n/db/src/repositories/index.ts @@ -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, diff --git a/packages/@n8n/engine/src/execution/execution-view-store.ts b/packages/@n8n/engine/src/execution/execution-view-store.ts index 0c6dfa9bb36..7e437a4d574 100644 --- a/packages/@n8n/engine/src/execution/execution-view-store.ts +++ b/packages/@n8n/engine/src/execution/execution-view-store.ts @@ -24,7 +24,7 @@ export interface ExecutionListItemView { } export interface ExecutionListQuery { - workflowIds: string[] | 'all'; + workflowIds: readonly string[] | 'all'; status?: ExecutionStatus[]; mode?: string; createdAfter?: string; diff --git a/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts b/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts index 6f78f270381..1aaa8ddfba9 100644 --- a/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts +++ b/packages/cli/src/executions/__tests__/engine-v2-execution-reader.service.test.ts @@ -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', () => { diff --git a/packages/cli/src/executions/__tests__/execution-cursor.test.ts b/packages/cli/src/executions/__tests__/execution-cursor.test.ts index 5788acb3693..fdf756b4012 100644 --- a/packages/cli/src/executions/__tests__/execution-cursor.test.ts +++ b/packages/cli/src/executions/__tests__/execution-cursor.test.ts @@ -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'); }); }); diff --git a/packages/cli/src/executions/__tests__/execution-list-v1.service.test.ts b/packages/cli/src/executions/__tests__/execution-list-v1.service.test.ts index 7fd96337b4b..9728532ab7e 100644 --- a/packages/cli/src/executions/__tests__/execution-list-v1.service.test.ts +++ b/packages/cli/src/executions/__tests__/execution-list-v1.service.test.ts @@ -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({ type: 'sqlite' }); + const databaseConfig = mock(); const executionRepository = mock(); - 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({ id, status, startedAt, createdAt: startedAt }); + + const query = ( + overrides: Partial = {}, + ): 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 = {}) => + 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({ id: '2' }), - mock({ 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({ 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({ id: '20' })] - : [mock({ 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, + }); }); }); }); diff --git a/packages/cli/src/executions/__tests__/execution-list-v2.service.test.ts b/packages/cli/src/executions/__tests__/execution-list-v2.service.test.ts new file mode 100644 index 00000000000..978e4ffc88e --- /dev/null +++ b/packages/cli/src/executions/__tests__/execution-list-v2.service.test.ts @@ -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(); + const workflows = mock(); + const dataPlane = mock(); + const databaseConfig = mock(); + 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({ id: 'member', role: { slug: 'global:member', scopes: [] } }); + const owner = mock({ + id: 'owner', + role: { slug: 'global:owner', scopes: [{ slug: 'workflow:read' }] }, + }); + const query = ( + overrides: Partial = {}, + ): 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 = {}) => + 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({ + 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> = [ + { 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', + ); + }); + }); +}); diff --git a/packages/cli/src/executions/__tests__/execution-list.service.test.ts b/packages/cli/src/executions/__tests__/execution-list.service.test.ts index 5ffad5fcb21..5fff8dbace9 100644 --- a/packages/cli/src/executions/__tests__/execution-list.service.test.ts +++ b/packages/cli/src/executions/__tests__/execution-list.service.test.ts @@ -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(); + const v2Service = mock(); + const dataPlane = mock(); const workflowSharingService = mock(); const roleService = mock(); + const service = new ExecutionListService( + v1Service, + dataPlane, + workflowSharingService, + roleService, + ); - const service = new ExecutionListService(v1Service, workflowSharingService, roleService); + const query = ( + overrides: Partial = {}, + ): ExecutionSummaries.RangeQuery => ({ + kind: 'range', + range: { limit: 20 }, + ...overrides, + }); + + /** The `query` above, as `listExecutionsForUI` narrows it when the two blocks combine. */ + const completedPageQuery = (overrides: Partial = {}) => ({ + ...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({ workflowId: 'wf-1' }), - mock({ 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({ 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'], + [], + ]); }); }); }); diff --git a/packages/cli/src/executions/__tests__/execution.service.test.ts b/packages/cli/src/executions/__tests__/execution.service.test.ts index 597d2cd22c9..ae65f5ca20a 100644 --- a/packages/cli/src/executions/__tests__/execution.service.test.ts +++ b/packages/cli/src/executions/__tests__/execution.service.test.ts @@ -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'; diff --git a/packages/cli/src/executions/__tests__/executions.controller.test.ts b/packages/cli/src/executions/__tests__/executions.controller.test.ts index e88465bc4c6..ca7b4e616ab 100644 --- a/packages/cli/src/executions/__tests__/executions.controller.test.ts +++ b/packages/cli/src/executions/__tests__/executions.controller.test.ts @@ -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(); const executionListService = mock(); const workflowSharingService = mock(); - const license = mock(); 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({ 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({ 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({ 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({ 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', () => { diff --git a/packages/cli/src/executions/__tests__/merge-execution-pages.test.ts b/packages/cli/src/executions/__tests__/merge-execution-pages.test.ts new file mode 100644 index 00000000000..7db5b36d390 --- /dev/null +++ b/packages/cli/src/executions/__tests__/merge-execution-pages.test.ts @@ -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(); + }); +}); diff --git a/packages/cli/src/executions/__tests__/parse-range-query.middleware.test.ts b/packages/cli/src/executions/__tests__/parse-range-query.middleware.test.ts index 01c336f479d..6b7e36b7694 100644 --- a/packages/cli/src/executions/__tests__/parse-range-query.middleware.test.ts +++ b/packages/cli/src/executions/__tests__/parse-range-query.middleware.test.ts @@ -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) => + mock({ + 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({ - 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({ - 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({ - 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({ - 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({ - 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({ - 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({ - 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({ - 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({ - 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({ - 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({ - 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({ - 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({ - 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); diff --git a/packages/cli/src/executions/engine-v2-execution-reader.service.ts b/packages/cli/src/executions/engine-v2-execution-reader.service.ts index 07e40029d31..9da3c9ed5bb 100644 --- a/packages/cli/src/executions/engine-v2-execution-reader.service.ts +++ b/packages/cli/src/executions/engine-v2-execution-reader.service.ts @@ -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([ - ['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; -/** Anything not manual is a production run. */ -const MODE_V1 = new Map([ - ['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([ 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, diff --git a/packages/cli/src/executions/engine-v2-mapping.ts b/packages/cli/src/executions/engine-v2-mapping.ts new file mode 100644 index 00000000000..6e69d5626ba --- /dev/null +++ b/packages/cli/src/executions/engine-v2-mapping.ts @@ -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([ + ['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([ + ['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); +} diff --git a/packages/cli/src/executions/execution-cursor.ts b/packages/cli/src/executions/execution-cursor.ts index df9445bbb51..30d8fd7660b 100644 --- a/packages/cli/src/executions/execution-cursor.ts +++ b/packages/cli/src/executions/execution-cursor.ts @@ -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; +export type ExecutionPosition = NonNullable; -/** 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; +/** 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; } diff --git a/packages/cli/src/executions/execution-list-v1.service.ts b/packages/cli/src/executions/execution-list-v1.service.ts index 03e49645514..a12d2a7f852 100644 --- a/packages/cli/src/executions/execution-list-v1.service.ts +++ b/packages/cli/src/executions/execution-list-v1.service.ts @@ -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 { - const { range: _, ...countQuery } = query; + async findPageWithCount( + query: ExecutionSummaries.RangeQuery, + cursor?: ExecutionCursor, + ): Promise { + 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(), + }, + }); } /** diff --git a/packages/cli/src/executions/execution-list-v2.service.ts b/packages/cli/src/executions/execution-list-v2.service.ts new file mode 100644 index 00000000000..08c92fb4746 --- /dev/null +++ b/packages/cli/src/executions/execution-list-v2.service.ts @@ -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, + readonly v2Reader: EngineV2ExecutionReader, + /** `null` keeps the data plane out of this request. */ + readonly v2Scopes: V2Scope | null, + readonly cursor?: Readonly, + ) {} +} + +/** 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 { + 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 { + 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 { + 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 { + 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 { + 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); + } +} diff --git a/packages/cli/src/executions/execution-list.service.ts b/packages/cli/src/executions/execution-list.service.ts index 7bd40817064..d2fc9933be0 100644 --- a/packages/cli/src/executions/execution-list.service.ts +++ b/packages/cli/src/executions/execution-list.service.ts @@ -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 { - // 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 { + // 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 { - return await this.v1Service.findPageWithCount(query); + async findPageWithCount( + query: ExecutionSummaries.RangeQuery, + cursor?: ExecutionCursor, + ): Promise { + 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 { + if (!this.dataPlane.isAvailable()) return this.v1Service; + + const { ExecutionListV2Service } = await import('./execution-list-v2.service.js'); + + return Container.get(ExecutionListV2Service); + } } diff --git a/packages/cli/src/executions/execution.types.ts b/packages/cli/src/executions/execution.types.ts index 7190b4bcc00..01831ff21a2 100644 --- a/packages/cli/src/executions/execution.types.ts +++ b/packages/cli/src/executions/execution.types.ts @@ -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; diff --git a/packages/cli/src/executions/executions.controller.ts b/packages/cli/src/executions/executions.controller.ts index 136ad42d8f5..d482500060b 100644 --- a/packages/cli/src/executions/executions.controller.ts +++ b/packages/cli/src/executions/executions.controller.ts @@ -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( diff --git a/packages/cli/src/executions/merge-execution-pages.ts b/packages/cli/src/executions/merge-execution-pages.ts new file mode 100644 index 00000000000..bcd9bda4242 --- /dev/null +++ b/packages/cli/src/executions/merge-execution-pages.ts @@ -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, + }; +} diff --git a/packages/cli/src/executions/parse-range-query.middleware.ts b/packages/cli/src/executions/parse-range-query.middleware.ts index 58a49bd1f2b..86353f42c00 100644 --- a/packages/cli/src/executions/parse-range-query.middleware.ts +++ b/packages/cli/src/executions/parse-range-query.middleware.ts @@ -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) { diff --git a/packages/cli/src/modules/engine-v2/__tests__/engine-data-plane-client.test.ts b/packages/cli/src/modules/engine-v2/__tests__/engine-data-plane-client.test.ts index 9ca0d114b48..611f68132c1 100644 --- a/packages/cli/src/modules/engine-v2/__tests__/engine-data-plane-client.test.ts +++ b/packages/cli/src/modules/engine-v2/__tests__/engine-data-plane-client.test.ts @@ -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' }); diff --git a/packages/cli/src/modules/engine-v2/engine-data-plane-client.ts b/packages/cli/src/modules/engine-v2/engine-data-plane-client.ts index 75501695955..e48cae3dd00 100644 --- a/packages/cli/src/modules/engine-v2/engine-data-plane-client.ts +++ b/packages/cli/src/modules/engine-v2/engine-data-plane-client.ts @@ -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 { + const response = await this.http.request({ + 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; diff --git a/packages/cli/src/modules/mcp/__tests__/search-executions.tool.test.ts b/packages/cli/src/modules/mcp/__tests__/search-executions.tool.test.ts index b90dc457c56..bec1d288817 100644 --- a/packages/cli/src/modules/mcp/__tests__/search-executions.tool.test.ts +++ b/packages/cli/src/modules/mcp/__tests__/search-executions.tool.test.ts @@ -26,23 +26,25 @@ const createExecution = (overrides: Partial = {}): 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'); diff --git a/packages/cli/src/modules/mcp/mcp.service.ts b/packages/cli/src/modules/mcp/mcp.service.ts index 8f9289e27fd..effc9ef2955 100644 --- a/packages/cli/src/modules/mcp/mcp.service.ts +++ b/packages/cli/src/modules/mcp/mcp.service.ts @@ -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, diff --git a/packages/cli/src/modules/mcp/tools/search-executions.tool.ts b/packages/cli/src/modules/mcp/tools/search-executions.tool.ts index 5f3f1fabce6..8b47b195ddb 100644 --- a/packages/cli/src/modules/mcp/tools/search-executions.tool.ts +++ b/packages/cli/src/modules/mcp/tools/search-executions.tool.ts @@ -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, diff --git a/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts b/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts index 98a8306005a..11e5d45b098 100644 --- a/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts +++ b/packages/cli/src/services/__tests__/engine-data-plane-proxy.service.test.ts @@ -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(); + 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(); const snapshot = mock({ id: executionId }); diff --git a/packages/cli/src/services/engine-data-plane-proxy.service.ts b/packages/cli/src/services/engine-data-plane-proxy.service.ts index 2ed64a50f72..d2e37cd7721 100644 --- a/packages/cli/src/services/engine-data-plane-proxy.service.ts +++ b/packages/cli/src/services/engine-data-plane-proxy.service.ts @@ -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; startExecution(request: StartExecutionRequest): Promise; /** @@ -44,6 +51,11 @@ export class EngineDataPlaneProxyService implements EngineDataPlaneProvider { return this.provider !== null; } + async searchExecutions(request: SearchExecutionsRequest): Promise { + if (!this.provider) return { items: [], nextCursor: null, total: 0 }; + return await this.provider.searchExecutions(request); + } + async startExecution(request: StartExecutionRequest): Promise { if (!this.provider) { throw new UserError( diff --git a/packages/cli/test/integration/engine-v2-active-triggers.test.ts b/packages/cli/test/integration/engine-v2-active-triggers.test.ts index c8d5cce8bd4..bba2bacaa8c 100644 --- a/packages/cli/test/integration/engine-v2-active-triggers.test.ts +++ b/packages/cli/test/integration/engine-v2-active-triggers.test.ts @@ -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 () => { diff --git a/packages/cli/test/integration/engine-v2-webhooks.test.ts b/packages/cli/test/integration/engine-v2-webhooks.test.ts index 793da9ffc8f..b40dfd64fee 100644 --- a/packages/cli/test/integration/engine-v2-webhooks.test.ts +++ b/packages/cli/test/integration/engine-v2-webhooks.test.ts @@ -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 { diff --git a/packages/cli/test/integration/execution.service.integration.test.ts b/packages/cli/test/integration/execution.service.integration.test.ts index b73b43574af..88dd4a560c3 100644 --- a/packages/cli/test/integration/execution.service.integration.test.ts +++ b/packages/cli/test/integration/execution.service.integration.test.ts @@ -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, }); diff --git a/packages/cli/test/integration/executions.controller.test.ts b/packages/cli/test/integration/executions.controller.test.ts index 487a490aeb4..c2a0f4da88b 100644 --- a/packages/cli/test/integration/executions.controller.test.ts +++ b/packages/cli/test/integration/executions.controller.test.ts @@ -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(() => { diff --git a/packages/cli/test/integration/executions/execution-list.test.ts b/packages/cli/test/integration/executions/execution-list.test.ts new file mode 100644 index 00000000000..2386e52b5d1 --- /dev/null +++ b/packages/cli/test/integration/executions/execution-list.test.ts @@ -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(); +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); + }); +}); diff --git a/packages/cli/test/integration/workflows/workflows.controller.test.ts b/packages/cli/test/integration/workflows/workflows.controller.test.ts index e4f2c824d35..3ab4eb141e9 100644 --- a/packages/cli/test/integration/workflows/workflows.controller.test.ts +++ b/packages/cli/test/integration/workflows/workflows.controller.test.ts @@ -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(() => { diff --git a/packages/workflow/src/interfaces.ts b/packages/workflow/src/interfaces.ts index f001c79f133..03dc51d5be9 100644 --- a/packages/workflow/src/interfaces.ts +++ b/packages/workflow/src/interfaces.ts @@ -4255,7 +4255,7 @@ export interface ExecutionSummary { }; usedPrivateCredentials?: boolean; annotation?: { - vote: AnnotationVote; + vote?: AnnotationVote | null; tags: Array<{ id: string; name: string;