diff --git a/packages/@n8n/db/src/repositories/workflow-history.repository.ts b/packages/@n8n/db/src/repositories/workflow-history.repository.ts index fdd33d94b33..c1fa43def57 100644 --- a/packages/@n8n/db/src/repositories/workflow-history.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-history.repository.ts @@ -178,7 +178,9 @@ export class WorkflowHistoryRepository extends BaseRepository { metaData, ); - await this.delete({ versionId: In(grouped.removed.map((x) => x.versionId)) }); - return { seen: workflows.length, deleted: grouped.removed.length }; + const { affected } = await this.delete({ + versionId: In(grouped.removed.map((x) => x.versionId)), + }); + return { seen: workflows.length, deleted: affected ?? grouped.removed.length }; } } diff --git a/packages/cli/src/services/pruning/__tests__/workflow-history-compaction-optimize.task.test.ts b/packages/cli/src/services/pruning/__tests__/workflow-history-compaction-optimize.task.test.ts index 1b2f5424391..43d9cac9233 100644 --- a/packages/cli/src/services/pruning/__tests__/workflow-history-compaction-optimize.task.test.ts +++ b/packages/cli/src/services/pruning/__tests__/workflow-history-compaction-optimize.task.test.ts @@ -18,7 +18,7 @@ describe('WorkflowHistoryCompactionOptimizeTask', () => { expect(task.name).toBe('workflow-history-compaction-optimize'); expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 3600 }); expect(task.effects).toBe('idempotent'); - expect(task.placement).toEqual({ scope: 'cluster', durable: false }); + expect(task.placement).toEqual({ scope: 'cluster', durable: true }); }); it('should optimize histories on run, handing the pass its abort signal', async () => { diff --git a/packages/cli/src/services/pruning/__tests__/workflow-history-compaction.service.test.ts b/packages/cli/src/services/pruning/__tests__/workflow-history-compaction.service.test.ts index d24b856d6d8..eed602cb9f5 100644 --- a/packages/cli/src/services/pruning/__tests__/workflow-history-compaction.service.test.ts +++ b/packages/cli/src/services/pruning/__tests__/workflow-history-compaction.service.test.ts @@ -234,13 +234,16 @@ describe('WorkflowHistoryCompactionService', () => { }); describe('compactHistories', () => { - const createService = (workflowHistoryRepository: WorkflowHistoryRepository) => { + const createService = ( + workflowHistoryRepository: WorkflowHistoryRepository, + isLeader = true, + ) => { const eventService = mock(); const compactingService = new WorkflowHistoryCompactionService( config, globalConfig, mockLogger(), - mock({ isLeader: true, instanceType: 'main', isMultiMain: true }), + mock({ isLeader, instanceType: 'main', isMultiMain: true }), dbConnection, workflowHistoryRepository, eventService, @@ -262,6 +265,17 @@ describe('WorkflowHistoryCompactionService', () => { ); }); + it('should optimize on a main that is not the leader', async () => { + const workflowHistoryRepository = mock(); + workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue(['wf-1']); + workflowHistoryRepository.pruneHistory.mockResolvedValue({ seen: 2, deleted: 1 }); + const { compactingService } = createService(workflowHistoryRepository, false); + + await compactingService.optimizeHistories(new AbortController().signal); + + expect(workflowHistoryRepository.pruneHistory).toHaveBeenCalledOnce(); + }); + it('should trim over the window between trimmingMinimumAge and the time window', async () => { const workflowHistoryRepository = mock(); workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue([]); diff --git a/packages/cli/src/services/pruning/workflow-history-compaction-optimize.task.ts b/packages/cli/src/services/pruning/workflow-history-compaction-optimize.task.ts index 4e57fbbb772..6af002c62c1 100644 --- a/packages/cli/src/services/pruning/workflow-history-compaction-optimize.task.ts +++ b/packages/cli/src/services/pruning/workflow-history-compaction-optimize.task.ts @@ -21,7 +21,7 @@ export class WorkflowHistoryCompactionOptimizeTask implements SystemTask { readonly effects: SystemTaskEffects = 'idempotent'; - readonly placement: SystemTaskPlacement = { scope: 'cluster', durable: false }; + readonly placement: SystemTaskPlacement = { scope: 'cluster', durable: true }; constructor( private readonly config: WorkflowHistoryCompactionConfig, diff --git a/packages/cli/test/integration/workflow-history-compaction-optimize.task.test.ts b/packages/cli/test/integration/workflow-history-compaction-optimize.task.test.ts new file mode 100644 index 00000000000..38308121b50 --- /dev/null +++ b/packages/cli/test/integration/workflow-history-compaction-optimize.task.test.ts @@ -0,0 +1,178 @@ +import type { EventService } from '@n8n/backend-services'; +import { createWorkflow, createWorkflowHistory, mockLogger, testDb } from '@n8n/backend-test-utils'; +import { GlobalConfig } from '@n8n/config'; +import { Time } from '@n8n/constants'; +import { DbConnection, WorkflowHistoryRepository } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; +import repeat from 'lodash/repeat'; +import type { InstanceSettings } from 'n8n-core'; +import type { INode } from 'n8n-workflow'; +import { v4 as uuid } from 'uuid'; +import { mock } from 'vitest-mock-extended'; + +import { WorkflowHistoryCompactionOptimizeTask } from '@/services/pruning/workflow-history-compaction-optimize.task'; +import { WorkflowHistoryCompactionService } from '@/services/pruning/workflow-history-compaction.service'; + +const OVERLAPPING_RUNS = 4; +const WORKFLOWS = 12; +const VERSIONS_PER_WORKFLOW = 8; +const PROTECTED_INDEX = 3; + +type VersionSeed = { + versionId: string; + offsetMs: number; + named: boolean; + published: boolean; + authors: string; +}; + +// Keep all versions inside the optimization window. Include names, publications, +// session gaps, and author changes that must retain intermediate versions. +const seeds: VersionSeed[][] = Array.from({ length: WORKFLOWS }, (_, workflowIndex) => + Array.from({ length: VERSIONS_PER_WORKFLOW }, (_, k) => { + const variant = workflowIndex % 4; + const sessionGapMs = variant === 3 && k >= 4 ? 30 * Time.minutes.toMilliseconds : 0; + return { + versionId: uuid(), + offsetMs: k * Time.minutes.toMilliseconds + sessionGapMs, + named: variant === 1 && k === PROTECTED_INDEX, + published: variant === 2 && k === PROTECTED_INDEX, + authors: variant === 3 && k >= 6 ? 'Other User' : 'Test User', + }; + }), +); + +const expectedVersionIds = seeds + .flatMap((versions, workflowIndex) => { + const retainedIndexes = [[7], [3, 7], [3, 7], [3, 5, 7]][workflowIndex % 4]; + return retainedIndexes.map((index) => versions[index].versionId); + }) + .sort(); + +describe('WorkflowHistoryCompactionOptimizeTask', () => { + const signal = new AbortController().signal; + let repository: WorkflowHistoryRepository; + const cleanups: Array<() => Promise> = []; + const releases: Array<() => void> = []; + + const node = { + id: uuid(), + name: 'node', + parameters: {}, + type: 'aNodeType', + typeVersion: 1, + position: [0, 0], + } satisfies INode; + + beforeAll(async () => { + // Allow concurrent queries even when the test runner defaults to one connection. + Container.get(GlobalConfig).database.postgresdb.poolSize = OVERLAPPING_RUNS; + await testDb.init(); + repository = Container.get(WorkflowHistoryRepository); + }); + + afterEach(async () => { + for (const release of releases.splice(0)) { + release(); + } + await Promise.all(cleanups.splice(0).map(async (cleanup) => await cleanup())); + vi.restoreAllMocks(); + await testDb.truncate(['WorkflowEntity', 'WorkflowHistory', 'WorkflowPublishHistory']); + }); + + afterAll(async () => await testDb.terminate()); + + // Each main has its own service, so the local guard cannot serialize these passes. + function createTask() { + const globalConfig = Container.get(GlobalConfig); + const config = { ...globalConfig.workflowHistoryCompaction, batchDelayMs: 0 }; + const logger = mockLogger(); + const eventService = mock(); + const service = new WorkflowHistoryCompactionService( + config, + globalConfig, + logger, + mock({ instanceType: 'main', isLeader: false }), + Container.get(DbConnection), + repository, + eventService, + ); + return { + task: new WorkflowHistoryCompactionOptimizeTask(config, service), + logger, + eventService, + }; + } + + async function seedHistories(): Promise { + const base = Date.now() - Time.hours.toMilliseconds; + for (const versions of seeds) { + const workflow = await createWorkflow({ versionId: versions[versions.length - 1].versionId }); + for (const [k, version] of versions.entries()) { + await createWorkflowHistory( + { + ...workflow, + versionId: version.versionId, + nodes: [{ ...node, parameters: { a: repeat('1', k + 1) } }], + }, + undefined, + version.published ? {} : undefined, + { + createdAt: new Date(base + version.offsetMs), + authors: version.authors, + name: version.named ? 'Named version' : null, + }, + ); + } + } + } + + async function remainingVersionIds(): Promise { + const rows = await repository.find({ select: { versionId: true } }); + return rows.map(({ versionId }) => versionId).sort(); + } + + it('leaves the same versions as one pass when passes on several mains overlap', async () => { + await seedHistories(); + await createTask().task.run(signal); + const afterOnePass = await remainingVersionIds(); + expect(afterOnePass).toEqual(expectedVersionIds); + await testDb.truncate(['WorkflowEntity', 'WorkflowHistory', 'WorkflowPublishHistory']); + + await seedHistories(); + const ready = createDeferredPromise(); + releases.push(() => ready.resolve()); + const deleteVersions = repository.delete.bind(repository); + let waiting = 0; + vi.spyOn(repository, 'delete').mockImplementation(async (criteria) => { + // All mains must read their first snapshot before any of them deletes it. + if (++waiting === OVERLAPPING_RUNS) { + ready.resolve(); + } + await ready.promise; + return await deleteVersions(criteria); + }); + const tasks = Array.from({ length: OVERLAPPING_RUNS }, createTask); + const passes = tasks.map(async ({ task }) => await task.run(signal)); + cleanups.push(async () => await Promise.allSettled(passes)); + await Promise.all(passes); + + expect(await remainingVersionIds()).toEqual(afterOnePass); + const reportedDeletes = tasks + .flatMap(({ eventService }) => eventService.emit.mock.calls) + .reduce( + (sum, [, payload]) => + sum + (payload as { totalVersionsDeleted: number }).totalVersionsDeleted, + 0, + ); + expect(reportedDeletes).toBe(WORKFLOWS * VERSIONS_PER_WORKFLOW - afterOnePass.length); + for (const { logger, eventService } of tasks) { + expect(logger.error).not.toHaveBeenCalled(); + expect(eventService.emit).toHaveBeenCalledWith( + 'history-compacted', + expect.objectContaining({ workflowsProcessed: WORKFLOWS, errorCount: 0 }), + ); + } + }); +});