From 38eea65993339ea3e8156cdbacef247749331351 Mon Sep 17 00:00:00 2001 From: Veljko Terzic Date: Tue, 6 Oct 2026 12:17:32 +0000 Subject: [PATCH] feat(core): Run the workflow history trim as a durable daily system task (#40085) --- .../workflow-history.repository.ts | 48 ++- .../workflow-publish-history.repository.ts | 11 - .../__tests__/main-system-tasks.test.ts | 12 +- .../__tests__/system-task-runner.test.ts | 19 +- .../system-tasks/main-system-tasks.ts | 12 +- ...kflow-history-compaction-trim.task.test.ts | 50 +-- ...orkflow-history-compaction.service.test.ts | 168 +++++----- .../workflow-history-compaction.utils.test.ts | 19 ++ .../workflow-history-compaction-trim.task.ts | 19 +- .../workflow-history-compaction.service.ts | 33 +- .../workflow-history-compaction.utils.ts | 18 ++ .../workflow-history.repository.test.ts | 290 ++++++++++++++---- ...kflow-history-compaction-trim.task.test.ts | 137 +++++++++ packages/workflow/src/workflow-diff.ts | 13 +- packages/workflow/test/workflow-diff.test.ts | 120 ++++++++ 15 files changed, 722 insertions(+), 247 deletions(-) create mode 100644 packages/cli/src/services/pruning/__tests__/workflow-history-compaction.utils.test.ts create mode 100644 packages/cli/src/services/pruning/workflow-history-compaction.utils.ts create mode 100644 packages/cli/test/integration/workflow-history-compaction-trim.task.test.ts diff --git a/packages/@n8n/db/src/repositories/workflow-history.repository.ts b/packages/@n8n/db/src/repositories/workflow-history.repository.ts index c1fa43def57..a05662b9ed0 100644 --- a/packages/@n8n/db/src/repositories/workflow-history.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-history.repository.ts @@ -1,10 +1,14 @@ import { Service } from '@n8n/di'; -import { DataSource, In, LessThan } from '@n8n/typeorm'; +import { DataSource, In, IsNull, LessThan } from '@n8n/typeorm'; import { DiffMetaData, DiffRule, groupWorkflows, SKIP_RULES } from 'n8n-workflow'; -import { WorkflowHistory, WorkflowEntity, WorkflowPublishedVersion } from '../entities'; +import { + WorkflowHistory, + WorkflowEntity, + WorkflowPublishedVersion, + WorkflowPublishHistory, +} from '../entities'; import { BaseRepository } from './base-repository'; -import { WorkflowPublishHistoryRepository } from './workflow-publish-history.repository'; import { WorkflowReviewRequestWorkflow } from '../entities/workflow-review-request-workflow.ee'; import { WorkflowReviewRequest } from '../entities/workflow-review-request.ee'; import type { OperationContext } from '../services/transaction'; @@ -12,11 +16,7 @@ import { TransactionRunner } from '../services/transaction'; @Service() export class WorkflowHistoryRepository extends BaseRepository { - constructor( - dataSource: DataSource, - private readonly workflowPublishHistoryRepository: WorkflowPublishHistoryRepository, - transactionRunner: TransactionRunner, - ) { + constructor(dataSource: DataSource, transactionRunner: TransactionRunner) { super(WorkflowHistory, dataSource.manager, transactionRunner); } @@ -150,8 +150,21 @@ export class WorkflowHistoryRepository extends BaseRepository { skipRules: DiffRule[] = [], metaData?: Partial>, ): Promise<{ seen: number; deleted: number }> { - const workflows = await this.manager + const publishedVersionSubquery = this.manager + .createQueryBuilder() + .subQuery() + .select('1') + .from(WorkflowPublishHistory, 'wph') + .where('wph.workflowId = wh.workflowId') + .andWhere('wph.versionId = wh.versionId') + .getQuery(); + const { entities: workflows, raw } = await this.manager .createQueryBuilder(WorkflowHistory, 'wh') + .leftJoin(WorkflowEntity, 'w', 'w.id = wh.workflowId') + .addSelect( + `CASE WHEN w.versionId = wh.versionId OR EXISTS ${publishedVersionSubquery} THEN 1 ELSE 0 END`, + 'isProtected', + ) .where('wh.workflowId = :workflowId', { workflowId }) .andWhere('wh.createdAt <= :endDate', { endDate, @@ -160,26 +173,29 @@ export class WorkflowHistoryRepository extends BaseRepository { startDate, }) .orderBy('wh.createdAt', 'ASC') - .getMany(); + .addOrderBy('wh.versionId', 'ASC') + .getRawAndEntities<{ wh_versionId: string; isProtected: number }>(); - // Group by workflowId - const publishedVersions = - await this.workflowPublishHistoryRepository.getPublishedVersions(workflowId); + // The current version and every version that was ever published stay. + const protectedVersions = new Set( + raw.filter((row) => Number(row.isProtected) === 1).map((row) => row.wh_versionId), + ); const grouped = groupWorkflows( workflows, rules, [ - this.makeSkipActiveAndNamedVersionsRule( - new Set(publishedVersions.map((v) => v.versionId).filter((v) => v !== null)), - ), + this.makeSkipActiveAndNamedVersionsRule(protectedVersions), SKIP_RULES.skipDifferentUsers, ...skipRules, ], metaData, ); + // A version named after the read above stays, like one named before it. const { affected } = await this.delete({ versionId: In(grouped.removed.map((x) => x.versionId)), + name: IsNull(), + description: IsNull(), }); return { seen: workflows.length, deleted: affected ?? grouped.removed.length }; } diff --git a/packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts b/packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts index d814db10aec..5b7a0fe702e 100644 --- a/packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts @@ -38,17 +38,6 @@ export class WorkflowPublishHistoryRepository extends Repository>> { - return await this.manager - .createQueryBuilder(WorkflowPublishHistory, 'wph') - .select('wph.versionId') - .distinct(true) - .where('wph.workflowId = :workflowId', { workflowId }) - .getMany(); - } - async findActivatedByUserId(workflowId: string): Promise { const record = await this.findOne({ select: ['userId'], diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts index ef2480f45db..99ebc81cdea 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts +++ b/packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts @@ -18,12 +18,15 @@ const configWith = ({ pruneData = true, useWorkflowPublicationService = true, diagnosticsEnabled = true, + historyPruneTimeHours = -1, dbType = 'postgresdb', } = {}) => mock({ executions: { pruneData }, workflows: { useWorkflowPublicationService }, diagnostics: { enabled: diagnosticsEnabled }, + workflowHistory: { pruneTime: historyPruneTimeHours }, + workflowHistoryCompaction: { trimmingMinimumAgeDays: 6 }, database: { type: dbType as GlobalConfig['database']['type'] }, }); @@ -34,9 +37,9 @@ it('should return every main task when all features are on', async () => { ActivityPruningTask, LicenseRenewalTask, WorkflowHistoryCompactionOptimizeTask, - WorkflowHistoryCompactionTrimTask, WorkflowHistoryPruningTask, PendingAuthorizationCleanupTask, + WorkflowHistoryCompactionTrimTask, ExecutionPruningSoftDeleteTask, TelemetryPulseTask, WorkflowPublicationOutboxCleanupTask, @@ -57,6 +60,13 @@ it('should leave out execution pruning soft delete when pruning is off', async ( expect(tasks).toContain(WorkflowPublicationOutboxCleanupTask); }); +it('should leave out the history trim when the prune horizon is shorter than the trim window', async () => { + const tasks = await mainSystemTasks(configWith({ historyPruneTimeHours: 24 })); + + expect(tasks).not.toContain(WorkflowHistoryCompactionTrimTask); + expect(tasks).toContain(WorkflowHistoryCompactionOptimizeTask); +}); + it('should leave out outbox cleanup when the publication service is off', async () => { const tasks = await mainSystemTasks(configWith({ useWorkflowPublicationService: false })); diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts index 203832773e3..85b44e00cc6 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts +++ b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts @@ -34,6 +34,7 @@ describe('SystemTaskRunner', () => { enabledForSystemTasks = false, instanceRole = 'leader' as InstanceSettings['instanceRole'], instanceType = 'main' as InstanceSettings['instanceType'], + timezone = 'UTC', } = {}) { const logger = mock(); const metadata = new SystemTaskMetadata(); @@ -54,7 +55,7 @@ describe('SystemTaskRunner', () => { jobRegistrar, systemTaskOwner, mock({ - generic: { timezone: 'UTC' }, + generic: { timezone }, scheduler: { enabledForSystemTasks }, }), instanceSettings, @@ -865,6 +866,22 @@ describe('SystemTaskRunner', () => { expect(dummy.runCount).toBe(1); }); + it('runs a durable cron task on its timer, in the instance timezone, while the scheduler is inactive', async () => { + dummy.placement = { scope: 'cluster', durable: true }; + dummy.schedule = { kind: 'cron', cronExpression: '0 3 * * *', timezone: null }; + const { runner, metadata, durableScheduler } = setup({ timezone: 'Europe/Berlin' }); + metadata.register(DummySystemTask); + + await initRunner(runner); + // 03:00 in Berlin is 02:00 UTC in January. + await vi.advanceTimersByTimeAsync(2 * Time.hours.toMilliseconds - 1); + expect(dummy.runCount).toBe(0); + await vi.advanceTimersByTimeAsync(1); + + expect(durableScheduler.registerTaskHandler).not.toHaveBeenCalled(); + expect(dummy.runCount).toBe(1); + }); + it('declares a task it hands to the durable scheduler to the job owner', async () => { dummy.placement = { scope: 'cluster', durable: true }; const { runner, metadata, systemTaskOwner } = setup(durably); diff --git a/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts b/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts index 4f0df3e7afd..f572995b18f 100644 --- a/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts +++ b/packages/cli/src/scheduling/system-tasks/main-system-tasks.ts @@ -2,6 +2,7 @@ import type { GlobalConfig } from '@n8n/config'; import type { SystemTaskClass } from '@n8n/decorators'; import { ActivityPruningTask } from '@/services/pruning/activity-pruning.task'; +import { isTrimmingEnabled } from '@/services/pruning/workflow-history-compaction.utils'; /** * Return the main command's own system tasks, owned by no backend module. @@ -12,9 +13,6 @@ export async function mainSystemTasks(globalConfig: GlobalConfig): Promise { - let compactionService = mock(); - let task = new WorkflowHistoryCompactionTrimTask(compactionService); + const compactionService = mock(); + const task = new WorkflowHistoryCompactionTrimTask(compactionService); - const setService = ({ trimmingEnabled = true } = {}) => { - Object.defineProperty(compactionService, 'isTrimmingEnabled', { value: trimmingEnabled }); - }; - - beforeEach(() => { - compactionService = mock(); - task = new WorkflowHistoryCompactionTrimTask(compactionService); - }); - - afterEach(() => { - vi.useRealTimers(); - }); - - it('should declare an hourly tick', () => { + it('should declare a daily cron in the instance timezone and run durably', () => { expect(task.name).toBe('workflow-history-compaction-trim'); - expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 3600 }); + expect(task.schedule).toEqual({ kind: 'cron', cronExpression: '0 3 * * *', timezone: null }); expect(task.effects).toBe('idempotent'); - expect(task.placement).toEqual({ scope: 'cluster', durable: false }); + expect(task.placement).toEqual({ scope: 'cluster', durable: true }); }); - it('should trim at 3am server time, handing the pass its abort signal', async () => { - vi.setSystemTime(new Date(2026, 10, 10, 3, 0, 0)); - setService(); + it('should keep a missed occurrence claimable for an hour', () => { + expect(task.misfireGraceSeconds).toBe(Time.hours.toSeconds); + }); + + it('should trim on run, handing the pass its abort signal', async () => { const { signal } = new AbortController(); await task.run(signal); expect(compactionService.trimLongRunningHistories).toHaveBeenCalledExactlyOnceWith(signal); }); - - it('should not trim outside of 3am server time', async () => { - vi.setSystemTime(new Date(2026, 10, 10, 5, 0, 0)); - setService(); - - await task.run(new AbortController().signal); - - expect(compactionService.trimLongRunningHistories).not.toHaveBeenCalled(); - }); - - it('should not trim when the prune horizon is shorter than the trim window', async () => { - vi.setSystemTime(new Date(2026, 10, 10, 3, 0, 0)); - setService({ trimmingEnabled: false }); - - await task.run(new AbortController().signal); - - expect(compactionService.trimLongRunningHistories).not.toHaveBeenCalled(); - }); }); 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 eed602cb9f5..1a32a3eff96 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 @@ -6,10 +6,7 @@ import type { DbConnection, WorkflowHistoryRepository } from '@n8n/db'; import type { InstanceSettings } from 'n8n-core'; import { mock } from 'vitest-mock-extended'; -import { - getCompactionWindowDeltas, - WorkflowHistoryCompactionService, -} from '../workflow-history-compaction.service'; +import { WorkflowHistoryCompactionService } from '../workflow-history-compaction.service'; describe('WorkflowHistoryCompactionService', () => { const dbConnection = mock({ @@ -26,37 +23,21 @@ describe('WorkflowHistoryCompactionService', () => { skipOnStartUp: false, }); const globalConfig = mock({ + generic: { timezone: 'Europe/Berlin' }, workflowHistory: { pruneTime: -1, }, }); beforeEach(() => { - // Set the system to a time that isn't 3 AM to avoid hitting the "trim once a day" window - const mockDate = new Date(2026, 10, 10, 1, 0, 0); - vi.setSystemTime(mockDate); + // The window tests compare exact dates derived from `Date.now()`. + vi.setSystemTime(new Date(2026, 10, 10, 1, 0, 0)); }); afterEach(() => { vi.useRealTimers(); }); - describe('getCompactionWindowDeltas', () => { - it('should offset the start delta by the time window and leave the end delta at the minimum age', () => { - expect(getCompactionWindowDeltas(24, 2, Time.hours.toMilliseconds)).toEqual({ - startDelta: 26 * Time.hours.toMilliseconds, - endDelta: 24 * Time.hours.toMilliseconds, - }); - }); - - it('should return equal deltas for an empty time window', () => { - expect(getCompactionWindowDeltas(7, 0, Time.days.toMilliseconds)).toEqual({ - startDelta: 7 * Time.days.toMilliseconds, - endDelta: 7 * Time.days.toMilliseconds, - }); - }); - }); - describe('init', () => { it('should start compacting on main instance that is the leader', () => { const compactingService = new WorkflowHistoryCompactionService( @@ -95,7 +76,7 @@ describe('WorkflowHistoryCompactionService', () => { it('should skip trimming if pruneTime < trimAge', () => { const compactingService = new WorkflowHistoryCompactionService( - config, + { ...config, trimOnStartUp: true }, { ...globalConfig, workflowHistory: { pruneTime: 24 } }, mockLogger(), mock({ isLeader: true, instanceType: 'main', isMultiMain: true }), @@ -110,7 +91,6 @@ describe('WorkflowHistoryCompactionService', () => { compactingService.runStartupCompaction(); - expect(compactingService.isTrimmingEnabled).toBe(false); expect(trimLongRunningHistoriesSpy).not.toBeCalled(); }); it('should not skip trimming if pruneTime > trimAge', () => { @@ -207,32 +187,6 @@ describe('WorkflowHistoryCompactionService', () => { expect(optimizeHistoriesSpy).toHaveBeenCalled(); }); - it('should trim if triggered at 3 AM with trimOnStartUp as false', () => { - const mockDate = new Date(2026, 10, 10, 3, 0, 0); - vi.setSystemTime(mockDate); - - const compactingService = new WorkflowHistoryCompactionService( - { ...config, trimOnStartUp: false }, - globalConfig, - mockLogger(), - mock({ isLeader: true, instanceType: 'main', isMultiMain: true }), - dbConnection, - mock(), - mock(), - ); - - vi.spyOn(compactingService, 'optimizeHistories').mockImplementation((() => {}) as never); - const trimLongRunningHistoriesSpy = vi - .spyOn(compactingService, 'trimLongRunningHistories') - .mockImplementation((() => {}) as never); - - compactingService.runStartupCompaction(); - - expect(trimLongRunningHistoriesSpy).toHaveBeenCalled(); - - vi.useRealTimers(); - }); - describe('compactHistories', () => { const createService = ( workflowHistoryRepository: WorkflowHistoryRepository, @@ -276,20 +230,98 @@ describe('WorkflowHistoryCompactionService', () => { expect(workflowHistoryRepository.pruneHistory).toHaveBeenCalledOnce(); }); - it('should trim over the window between trimmingMinimumAge and the time window', async () => { + it('should trim over the window between trimmingMinimumAge and the time window, anchored to the start of the day in the instance timezone', async () => { + // 03:00 UTC on 10 November is 04:00 in Berlin, whose day started at 23:00 UTC the day before. + vi.setSystemTime(new Date('2026-11-10T03:00:00.000Z')); + const startOfDay = new Date('2026-11-09T23:00:00.000Z'); const workflowHistoryRepository = mock(); workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue([]); const { compactingService } = createService(workflowHistoryRepository); await compactingService['trimLongRunningHistories'](new AbortController().signal); - const now = Date.now(); expect(workflowHistoryRepository.getWorkflowIdsInRange).toHaveBeenCalledWith( - new Date(now - 9 * Time.days.toMilliseconds), - new Date(now - 7 * Time.days.toMilliseconds), + new Date(startOfDay.getTime() - 9 * Time.days.toMilliseconds), + new Date(startOfDay.getTime() - 7 * Time.days.toMilliseconds), ); }); + it('should read the same trim window for every pass of one day', async () => { + const workflowHistoryRepository = mock(); + workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue([]); + const { compactingService } = createService(workflowHistoryRepository); + + vi.setSystemTime(new Date('2026-11-10T03:00:00.000Z')); + await compactingService['trimLongRunningHistories'](new AbortController().signal); + vi.setSystemTime(new Date('2026-11-10T22:59:59.999Z')); + await compactingService['trimLongRunningHistories'](new AbortController().signal); + + const [first, second] = workflowHistoryRepository.getWorkflowIdsInRange.mock.calls; + expect(second).toEqual(first); + }); + + it('should move the trim window by one day at midnight in the instance timezone', async () => { + const workflowHistoryRepository = mock(); + workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue([]); + const { compactingService } = createService(workflowHistoryRepository); + + vi.setSystemTime(new Date('2026-11-10T22:59:59.999Z')); + await compactingService['trimLongRunningHistories'](new AbortController().signal); + vi.setSystemTime(new Date('2026-11-10T23:00:00.000Z')); + await compactingService['trimLongRunningHistories'](new AbortController().signal); + + const [[firstStart, firstEnd], [secondStart, secondEnd]] = + workflowHistoryRepository.getWorkflowIdsInRange.mock.calls; + expect(secondStart.getTime() - firstStart.getTime()).toBe(Time.days.toMilliseconds); + expect(secondEnd.getTime() - firstEnd.getTime()).toBe(Time.days.toMilliseconds); + }); + + it.each([ + // At 03:00 UTC it is still the day before in New York. + ['America/New_York', '2026-11-10T03:00:00.000Z', '2026-11-09T05:00:00.000Z'], + // 25 October 2026 has 25 hours in Berlin; the day still starts at its own midnight. + ['Europe/Berlin', '2026-10-25T03:00:00.000Z', '2026-10-24T22:00:00.000Z'], + ])( + 'should anchor the trim window to the start of the day in %s', + async (timezone, now, startOfDay) => { + vi.setSystemTime(new Date(now)); + const workflowHistoryRepository = mock(); + workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue([]); + const compactingService = new WorkflowHistoryCompactionService( + config, + mock({ generic: { timezone }, workflowHistory: { pruneTime: -1 } }), + mockLogger(), + mock({ isLeader: true, instanceType: 'main', isMultiMain: true }), + dbConnection, + workflowHistoryRepository, + mock(), + ); + + await compactingService['trimLongRunningHistories'](new AbortController().signal); + + const anchorMs = new Date(startOfDay).getTime(); + expect(workflowHistoryRepository.getWorkflowIdsInRange).toHaveBeenCalledWith( + new Date(anchorMs - 9 * Time.days.toMilliseconds), + new Date(anchorMs - 7 * Time.days.toMilliseconds), + ); + }, + ); + + it('should keep the optimize window on the clock while the trim window is anchored', async () => { + vi.setSystemTime(new Date('2026-11-10T03:00:00.000Z')); + const workflowHistoryRepository = mock(); + workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue([]); + const { compactingService } = createService(workflowHistoryRepository); + + await compactingService['optimizeHistories'](new AbortController().signal); + await compactingService['trimLongRunningHistories'](new AbortController().signal); + + const [[, optimizeEnd], [, trimEnd]] = + workflowHistoryRepository.getWorkflowIdsInRange.mock.calls; + expect(optimizeEnd).toEqual(new Date('2026-11-09T03:00:00.000Z')); + expect(trimEnd).toEqual(new Date('2026-11-02T23:00:00.000Z')); + }); + it('should not emit telemetry when no workflows are in range', async () => { const workflowHistoryRepository = mock(); workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue([]); @@ -410,30 +442,4 @@ describe('WorkflowHistoryCompactionService', () => { expect(signal.aborted).toBe(true); }); }); - - it('should not trim if triggered outside of 3 AM with trimOnStartUp as false', () => { - const mockDate = new Date(2026, 10, 10, 5, 0, 0); - vi.setSystemTime(mockDate); - - const compactingService = new WorkflowHistoryCompactionService( - { ...config, trimOnStartUp: false }, - globalConfig, - mockLogger(), - mock({ isLeader: true, instanceType: 'main', isMultiMain: true }), - dbConnection, - mock(), - mock(), - ); - - vi.spyOn(compactingService, 'optimizeHistories').mockImplementation((() => {}) as never); - const trimLongRunningHistoriesSpy = vi - .spyOn(compactingService, 'trimLongRunningHistories') - .mockImplementation((() => {}) as never); - - compactingService.runStartupCompaction(); - - expect(trimLongRunningHistoriesSpy).not.toHaveBeenCalled(); - - vi.useRealTimers(); - }); }); diff --git a/packages/cli/src/services/pruning/__tests__/workflow-history-compaction.utils.test.ts b/packages/cli/src/services/pruning/__tests__/workflow-history-compaction.utils.test.ts new file mode 100644 index 00000000000..0534c598669 --- /dev/null +++ b/packages/cli/src/services/pruning/__tests__/workflow-history-compaction.utils.test.ts @@ -0,0 +1,19 @@ +import { Time } from '@n8n/constants'; + +import { getCompactionWindowDeltas } from '../workflow-history-compaction.utils'; + +describe('getCompactionWindowDeltas', () => { + it('should offset the start delta by the time window and leave the end delta at the minimum age', () => { + expect(getCompactionWindowDeltas(24, 2, Time.hours.toMilliseconds)).toEqual({ + startDelta: 26 * Time.hours.toMilliseconds, + endDelta: 24 * Time.hours.toMilliseconds, + }); + }); + + it('should return equal deltas for an empty time window', () => { + expect(getCompactionWindowDeltas(7, 0, Time.days.toMilliseconds)).toEqual({ + startDelta: 7 * Time.days.toMilliseconds, + endDelta: 7 * Time.days.toMilliseconds, + }); + }); +}); diff --git a/packages/cli/src/services/pruning/workflow-history-compaction-trim.task.ts b/packages/cli/src/services/pruning/workflow-history-compaction-trim.task.ts index f48bd44d821..0111b073b67 100644 --- a/packages/cli/src/services/pruning/workflow-history-compaction-trim.task.ts +++ b/packages/cli/src/services/pruning/workflow-history-compaction-trim.task.ts @@ -1,5 +1,5 @@ import { Time } from '@n8n/constants'; -import { intervalFromSeconds, SystemTask } from '@n8n/decorators'; +import { SystemTask } from '@n8n/decorators'; import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from '@n8n/decorators'; import { WorkflowHistoryCompactionService } from './workflow-history-compaction.service'; @@ -12,19 +12,24 @@ import { WorkflowHistoryCompactionService } from './workflow-history-compaction. export class WorkflowHistoryCompactionTrimTask implements SystemTask { readonly name = 'workflow-history-compaction-trim'; - // Hourly tick with a 3am gate in run() mirrors the legacy timer, which had - // to survive leader changes. CAT-4173 replaces this with a cron schedule. - readonly schedule: SystemTaskSchedule = intervalFromSeconds(Time.hours.toSeconds); + readonly schedule: SystemTaskSchedule = { + kind: 'cron', + cronExpression: '0 3 * * *', + // `null` resolves to `GENERIC_TIMEZONE` at run time. + timezone: null, + }; readonly effects: SystemTaskEffects = 'idempotent'; - readonly placement: SystemTaskPlacement = { scope: 'cluster', durable: false }; + // A late run on the same day reads the same window, so an hour carries the + // occurrence across a restart at 03:00 instead of losing the day's trim. + readonly misfireGraceSeconds = Time.hours.toSeconds; + + readonly placement: SystemTaskPlacement = { scope: 'cluster', durable: true }; constructor(private readonly compactionService: WorkflowHistoryCompactionService) {} async run(signal: AbortSignal): Promise { - if (!this.compactionService.isTrimmingEnabled) return; - if (new Date().getHours() !== 3) return; await this.compactionService.trimLongRunningHistories(signal); } } diff --git a/packages/cli/src/services/pruning/workflow-history-compaction.service.ts b/packages/cli/src/services/pruning/workflow-history-compaction.service.ts index d335c22250c..d215de08826 100644 --- a/packages/cli/src/services/pruning/workflow-history-compaction.service.ts +++ b/packages/cli/src/services/pruning/workflow-history-compaction.service.ts @@ -8,14 +8,13 @@ import { Service } from '@n8n/di'; import { InstanceSettings } from 'n8n-core'; import { ensureError } from '@n8n/utils/errors/ensure-error'; import { sleep } from '@n8n/utils/sleep'; +import { DateTime } from 'luxon'; import { DiffMetaData, DiffRule, RULES, SKIP_RULES } from 'n8n-workflow'; import { strict } from 'node:assert'; import { RelayEventMap } from '@/events/maps/relay.event-map'; -export function getCompactionWindowDeltas(minimumAge: number, timeWindow: number, unitMs: number) { - return { startDelta: (minimumAge + timeWindow) * unitMs, endDelta: minimumAge * unitMs }; -} +import { getCompactionWindowDeltas, isTrimmingEnabled } from './workflow-history-compaction.utils'; /** * Responsible for compacting auto saved workflow history entries in the database. @@ -77,15 +76,6 @@ export class WorkflowHistoryCompactionService { return this.instanceSettings.instanceType === 'main' && this.instanceSettings.isLeader; } - /** Whether trimming may run at all: a prune horizon shorter than the trim window makes trimming pointless. */ - get isTrimmingEnabled() { - return ( - this.globalConfig.workflowHistory.pruneTime === -1 || - this.globalConfig.workflowHistory.pruneTime * Time.hours.toMilliseconds >= - this.config.trimmingMinimumAgeDays * Time.days.toMilliseconds - ); - } - // One-shot catch-up pass on startup and on leader change, so a gap between // leaders is compacted without waiting a full task interval. Not `runOnTakeover` // on the tasks: `trimOnStartUp` forces a trim only here, and a task run cannot @@ -101,8 +91,10 @@ export class WorkflowHistoryCompactionService { void this.optimizeHistories(signal); - if (!this.isTrimmingEnabled) return; - if (this.config.trimOnStartUp || new Date().getHours() === 3) { + if ( + this.config.trimOnStartUp && + isTrimmingEnabled(this.globalConfig.workflowHistory, this.config) + ) { void this.trimLongRunningHistories(signal); } } @@ -134,6 +126,7 @@ export class WorkflowHistoryCompactionService { try { await this.compactHistories( + this.startOfToday(), startDelta, endDelta, [ @@ -175,6 +168,7 @@ export class WorkflowHistoryCompactionService { try { await this.compactHistories( + new Date(), startDelta, endDelta, [RULES.mergeAdditiveChanges], @@ -186,7 +180,14 @@ export class WorkflowHistoryCompactionService { } } + /** Every trim pass of one day reads the same window, so a repeated pass finds nothing new. */ + private startOfToday(): Date { + return DateTime.now().setZone(this.globalConfig.generic.timezone).startOf('day').toJSDate(); + } + + /** The window ends `endDeltaMs` before `anchor` and starts `startDeltaMs` before it. */ private async compactHistories( + anchor: Date, startDeltaMs: number, endDeltaMs: number, rules: DiffRule[], @@ -196,8 +197,8 @@ export class WorkflowHistoryCompactionService { ): Promise { const compactionStartTime = Date.now(); - const startDate = new Date(compactionStartTime - startDeltaMs); - const endDate = new Date(compactionStartTime - endDeltaMs); + const startDate = new Date(anchor.getTime() - startDeltaMs); + const endDate = new Date(anchor.getTime() - endDeltaMs); const startIso = startDate.toISOString(); const endIso = endDate.toISOString(); diff --git a/packages/cli/src/services/pruning/workflow-history-compaction.utils.ts b/packages/cli/src/services/pruning/workflow-history-compaction.utils.ts new file mode 100644 index 00000000000..a2569d98058 --- /dev/null +++ b/packages/cli/src/services/pruning/workflow-history-compaction.utils.ts @@ -0,0 +1,18 @@ +import type { GlobalConfig, WorkflowHistoryCompactionConfig } from '@n8n/config'; +import { Time } from '@n8n/constants'; + +export function getCompactionWindowDeltas(minimumAge: number, timeWindow: number, unitMs: number) { + return { startDelta: (minimumAge + timeWindow) * unitMs, endDelta: minimumAge * unitMs }; +} + +/** A prune horizon shorter than the trim window makes trimming pointless. */ +export function isTrimmingEnabled( + workflowHistory: GlobalConfig['workflowHistory'], + compaction: WorkflowHistoryCompactionConfig, +): boolean { + return ( + workflowHistory.pruneTime === -1 || + workflowHistory.pruneTime * Time.hours.toMilliseconds >= + compaction.trimmingMinimumAgeDays * Time.days.toMilliseconds + ); +} diff --git a/packages/cli/test/integration/database/repositories/workflow-history.repository.test.ts b/packages/cli/test/integration/database/repositories/workflow-history.repository.test.ts index 0b3764576f8..f2bddffcdb1 100644 --- a/packages/cli/test/integration/database/repositories/workflow-history.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/workflow-history.repository.test.ts @@ -2,7 +2,6 @@ import { createTeamProject, createWorkflow, createWorkflowHistory, - createWorkflowWithHistory, testDb, } from '@n8n/backend-test-utils'; import { @@ -27,6 +26,12 @@ describe('WorkflowHistoryRepository', () => { const alwaysMergeRule = () => true; + // Rows written in the same millisecond have no insertion order, so every version gets its own second. + const secondsAgo = (seconds: number) => new Date(Date.now() - seconds * 1_000); + + // A readable version id of the 36 characters a real one has. Postgres pads a shorter workflow versionId. + const vid = (prefix: string) => prefix.padEnd(36, '0'); + beforeAll(async () => { await testDb.init(); }); @@ -53,25 +58,31 @@ describe('WorkflowHistoryRepository', () => { const id1 = uuid(); const id2 = uuid(); - const workflow = await createWorkflowWithHistory({ - versionId: id1, - nodes: [{ ...testNode1, parameters: { a: 'a' } }], - }); - await createWorkflowHistory({ - ...workflow, - versionId: uuid(), - nodes: [{ ...testNode1, parameters: { a: 'ab' } }], - }); - await createWorkflowHistory({ - ...workflow, - versionId: uuid(), - nodes: [{ ...testNode1, parameters: { a: 'abc' } }], - }); - await createWorkflowHistory({ - ...workflow, - versionId: id2, - nodes: [{ ...testNode1, parameters: { a: 'abcd' } }], - }); + const workflow = await createWorkflow({ versionId: id2 }); + await createWorkflowHistory( + { ...workflow, versionId: id1, nodes: [{ ...testNode1, parameters: { a: 'a' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(4) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: uuid(), nodes: [{ ...testNode1, parameters: { a: 'ab' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(3) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: uuid(), nodes: [{ ...testNode1, parameters: { a: 'abc' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(2) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: id2, nodes: [{ ...testNode1, parameters: { a: 'abcd' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(1) }, + ); // ACT const repository = Container.get(WorkflowHistoryRepository); @@ -128,25 +139,31 @@ describe('WorkflowHistoryRepository', () => { const id1 = uuid(); const id2 = uuid(); - const workflow = await createWorkflowWithHistory({ - versionId: id1, - nodes: [{ ...testNode1, parameters: { a: 'abcde' } }], - }); - await createWorkflowHistory({ - ...workflow, - versionId: uuid(), - nodes: [{ ...testNode1, parameters: { a: 'ab' } }], - }); - await createWorkflowHistory({ - ...workflow, - versionId: uuid(), - nodes: [{ ...testNode1, parameters: { a: 'abc' } }], - }); - await createWorkflowHistory({ - ...workflow, - versionId: id2, - nodes: [{ ...testNode1, parameters: { a: 'abcd' } }], - }); + const workflow = await createWorkflow({ versionId: id2 }); + await createWorkflowHistory( + { ...workflow, versionId: id1, nodes: [{ ...testNode1, parameters: { a: 'abcde' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(4) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: uuid(), nodes: [{ ...testNode1, parameters: { a: 'ab' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(3) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: uuid(), nodes: [{ ...testNode1, parameters: { a: 'abc' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(2) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: id2, nodes: [{ ...testNode1, parameters: { a: 'abcd' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(1) }, + ); // ACT const repository = Container.get(WorkflowHistoryRepository); @@ -187,39 +204,37 @@ describe('WorkflowHistoryRepository', () => { const id4 = uuid(); const id5 = uuid(); - const workflow = await createWorkflowWithHistory({ - versionId: id1, - nodes: [{ ...testNode1, parameters: { a: 'a' } }], - }); + const workflow = await createWorkflow({ versionId: id5 }); await createWorkflowHistory( - { - ...workflow, - versionId: id2, - nodes: [{ ...testNode1, parameters: { a: 'ab' } }], - }, + { ...workflow, versionId: id1, nodes: [{ ...testNode1, parameters: { a: 'a' } }] }, undefined, undefined, - { name: 'aVersionName' }, + { createdAt: secondsAgo(5) }, ); - await createWorkflowHistory({ - ...workflow, - versionId: id3, - nodes: [{ ...testNode1, parameters: { a: 'abc' } }], - }); await createWorkflowHistory( - { - ...workflow, - versionId: id4, - nodes: [{ ...testNode1, parameters: { a: 'abcd' } }], - }, + { ...workflow, versionId: id2, nodes: [{ ...testNode1, parameters: { a: 'ab' } }] }, + undefined, + undefined, + { name: 'aVersionName', createdAt: secondsAgo(4) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: id3, nodes: [{ ...testNode1, parameters: { a: 'abc' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(3) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: id4, nodes: [{ ...testNode1, parameters: { a: 'abcd' } }] }, undefined, { event: 'activated' }, + { createdAt: secondsAgo(2) }, + ); + await createWorkflowHistory( + { ...workflow, versionId: id5, nodes: [{ ...testNode1, parameters: { a: 'abcde' } }] }, + undefined, + undefined, + { createdAt: secondsAgo(1) }, ); - await createWorkflowHistory({ - ...workflow, - versionId: id5, - nodes: [{ ...testNode1, parameters: { a: 'abcde' } }], - }); // ACT const repository = Container.get(WorkflowHistoryRepository); @@ -251,6 +266,153 @@ describe('WorkflowHistoryRepository', () => { expect(redo.deleted).toBe(0); expect(redo.seen).toBe(3); }); + + it('should not delete a version that was named after the versions were read', async () => { + const id1 = uuid(); + const id2 = uuid(); + const workflow = await createWorkflow({ versionId: id2, nodes: [testNode1] }); + await createWorkflowHistory({ ...workflow, versionId: id1 }, undefined, undefined, { + createdAt: secondsAgo(2), + }); + await createWorkflowHistory({ ...workflow, versionId: id2 }, undefined, undefined, { + createdAt: secondsAgo(1), + }); + const repository = Container.get(WorkflowHistoryRepository); + const originalDelete = repository.delete.bind(repository); + const deleteSpy = vi + .spyOn(repository, 'delete') + .mockImplementationOnce(async (criteria: Parameters[0]) => { + await repository.update({ versionId: id1 }, { name: 'named late' }); + return await originalDelete(criteria); + }); + + const aDayAgo = new Date(Date.now() - 24 * 60 * 60_000); + const nextDay = new Date(Date.now() + 24 * 60 * 60_000); + const { deleted } = await repository.pruneHistory(workflow.id, aDayAgo, nextDay, [ + alwaysMergeRule, + ]); + deleteSpy.mockRestore(); + + expect(deleted).toBe(0); + const history = await repository.find({ order: { createdAt: 'ASC' } }); + expect(history).toEqual([ + expect.objectContaining({ versionId: id1, name: 'named late' }), + expect.objectContaining({ versionId: id2 }), + ]); + }); + + it('should not delete a version that was described after the versions were read', async () => { + const id1 = uuid(); + const id2 = uuid(); + const workflow = await createWorkflow({ versionId: id2, nodes: [testNode1] }); + await createWorkflowHistory({ ...workflow, versionId: id1 }, undefined, undefined, { + createdAt: secondsAgo(2), + }); + await createWorkflowHistory({ ...workflow, versionId: id2 }, undefined, undefined, { + createdAt: secondsAgo(1), + }); + const repository = Container.get(WorkflowHistoryRepository); + const originalDelete = repository.delete.bind(repository); + const deleteSpy = vi + .spyOn(repository, 'delete') + .mockImplementationOnce(async (criteria: Parameters[0]) => { + await repository.update({ versionId: id1 }, { description: 'described late' }); + return await originalDelete(criteria); + }); + + const { deleted, seen } = await repository.pruneHistory( + workflow.id, + secondsAgo(60), + new Date(), + [alwaysMergeRule], + ); + deleteSpy.mockRestore(); + + expect({ seen, deleted }).toEqual({ seen: 2, deleted: 0 }); + await expect(repository.count()).resolves.toBe(2); + }); + + it('should keep the greatest versionId of versions with the same createdAt, whatever their insertion order', async () => { + const createdAt = secondsAgo(30); + const workflow = await createWorkflow({ versionId: vid('v-c'), nodes: [testNode1] }); + for (const versionId of [vid('v-b'), vid('v-c'), vid('v-a')]) { + await createWorkflowHistory({ ...workflow, versionId }, undefined, undefined, { + createdAt, + }); + } + const repository = Container.get(WorkflowHistoryRepository); + + const { deleted } = await repository.pruneHistory(workflow.id, secondsAgo(60), new Date(), [ + alwaysMergeRule, + ]); + + expect(deleted).toBe(2); + const history = await repository.find(); + expect(history).toEqual([expect.objectContaining({ versionId: vid('v-c') })]); + }); + + it('should keep the current version of the workflow when another version shares its createdAt and sorts after it', async () => { + const createdAt = secondsAgo(30); + const workflow = await createWorkflow({ versionId: vid('v-a'), nodes: [testNode1] }); + for (const versionId of [vid('v-a'), vid('v-b')]) { + await createWorkflowHistory({ ...workflow, versionId }, undefined, undefined, { + createdAt, + }); + } + const repository = Container.get(WorkflowHistoryRepository); + + const { deleted } = await repository.pruneHistory(workflow.id, secondsAgo(60), new Date(), [ + alwaysMergeRule, + ]); + + expect(deleted).toBe(0); + const remaining = (await repository.find()).map((v) => v.versionId).sort(); + expect(remaining).toEqual([vid('v-a'), vid('v-b')]); + }); + + it('should keep the current version of the workflow when a newer unnamed version follows it', async () => { + const workflow = await createWorkflow({ versionId: vid('v-current'), nodes: [testNode1] }); + await createWorkflowHistory( + { ...workflow, versionId: vid('v-current') }, + undefined, + undefined, + { + createdAt: secondsAgo(2), + }, + ); + await createWorkflowHistory( + { ...workflow, versionId: vid('v-later') }, + undefined, + undefined, + { + createdAt: secondsAgo(1), + }, + ); + const repository = Container.get(WorkflowHistoryRepository); + + const { deleted } = await repository.pruneHistory(workflow.id, secondsAgo(60), new Date(), [ + alwaysMergeRule, + ]); + + expect(deleted).toBe(0); + await expect(repository.count()).resolves.toBe(2); + }); + + it('should order a version with the same createdAt before a newer one, whatever its versionId', async () => { + const workflow = await createWorkflow({ versionId: vid('v-a'), nodes: [testNode1] }); + await createWorkflowHistory({ ...workflow, versionId: vid('v-z') }, undefined, undefined, { + createdAt: secondsAgo(2), + }); + await createWorkflowHistory({ ...workflow, versionId: vid('v-a') }, undefined, undefined, { + createdAt: secondsAgo(1), + }); + const repository = Container.get(WorkflowHistoryRepository); + + await repository.pruneHistory(workflow.id, secondsAgo(60), new Date(), [alwaysMergeRule]); + + const history = await repository.find(); + expect(history).toEqual([expect.objectContaining({ versionId: vid('v-a') })]); + }); }); describe('deleteEarlierThanExceptCurrentAndActive', () => { // Happy path: proves the setup deletes old, unreferenced versions while diff --git a/packages/cli/test/integration/workflow-history-compaction-trim.task.test.ts b/packages/cli/test/integration/workflow-history-compaction-trim.task.test.ts new file mode 100644 index 00000000000..d4a1f080750 --- /dev/null +++ b/packages/cli/test/integration/workflow-history-compaction-trim.task.test.ts @@ -0,0 +1,137 @@ +import type { Logger } from '@n8n/backend-common'; +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 { DateTime } from 'luxon'; +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 { WorkflowHistoryCompactionTrimTask } from '@/services/pruning/workflow-history-compaction-trim.task'; +import { WorkflowHistoryCompactionService } from '@/services/pruning/workflow-history-compaction.service'; + +const OVERLAPPING_RUNS = 3; +// Inside the trim window, which is 6 to 8 days ago with the default config. +const VERSION_AGE_MS = 7 * Time.days.toMilliseconds; + +// One workflow whose versions are not evenly spaced and not the same size. A +// small workflow keeps one version per minute, measured to the kept neighbour. +const OFFSETS_S = [0, 100, 200, 280, 300, 380, 400]; +const LARGE_INDEX = 2; +const RETAINED_INDEXES = [0, 1, 2, 4, 6]; +const versionIds = OFFSETS_S.map(() => uuid()); +const expectedVersionIds = RETAINED_INDEXES.map((index) => versionIds[index]).sort(); + +const node = { + id: uuid(), + name: 'node', + parameters: { a: 'x' }, + type: 'aNodeType', + typeVersion: 1, + position: [0, 0], +} satisfies INode; + +describe('WorkflowHistoryCompactionTrimTask', () => { + const signal = new AbortController().signal; + let repository: WorkflowHistoryRepository; + + 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 () => { + vi.restoreAllMocks(); + await testDb.truncate(['WorkflowEntity', 'WorkflowHistory']); + }); + + 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 logger = mockLogger(); + const service = new WorkflowHistoryCompactionService( + { ...globalConfig.workflowHistoryCompaction, batchDelayMs: 0 }, + globalConfig, + logger, + mock({ instanceType: 'main', isLeader: false }), + Container.get(DbConnection), + repository, + mock(), + ); + return { task: new WorkflowHistoryCompactionTrimTask(service), logger }; + } + + async function seedHistory(): Promise { + // Keep all versions inside the day-anchored trim window, even near midnight. + const startOfDay = DateTime.now() + .setZone(Container.get(GlobalConfig).generic.timezone) + .startOf('day'); + const base = startOfDay.toMillis() - VERSION_AGE_MS; + const workflow = await createWorkflow({ + versionId: versionIds[versionIds.length - 1], + nodes: [node], + }); + for (const [k, versionId] of versionIds.entries()) { + const parameters = k === LARGE_INDEX ? { a: 'x'.repeat(200) } : node.parameters; + await createWorkflowHistory( + { ...workflow, versionId, nodes: [{ ...node, parameters }] }, + undefined, + undefined, + { createdAt: new Date(base + OFFSETS_S[k] * Time.seconds.toMilliseconds) }, + ); + } + } + + async function remainingVersionIds(): Promise { + const rows = await repository.find({ select: { versionId: true } }); + return rows.map(({ versionId }) => versionId).sort(); + } + + // Every pass reads its snapshot before any pass deletes from it. + async function runOverlappingPasses(): Promise { + const ready = createDeferredPromise(); + const deleteVersions = repository.delete.bind(repository); + let waiting = 0; + vi.spyOn(repository, 'delete').mockImplementation(async (criteria) => { + if (++waiting === OVERLAPPING_RUNS) ready.resolve(); + await ready.promise; + return await deleteVersions(criteria); + }); + const tasks = Array.from({ length: OVERLAPPING_RUNS }, createTask); + await Promise.all(tasks.map(async ({ task }) => await task.run(signal))); + return tasks.map(({ logger }) => logger); + } + + it('leaves the same versions as one pass when passes read the same snapshot', async () => { + await seedHistory(); + + const loggers = await runOverlappingPasses(); + + expect(await remainingVersionIds()).toEqual(expectedVersionIds); + for (const logger of loggers) { + expect(logger.error).not.toHaveBeenCalled(); + } + }); + + it('leaves the same versions as one pass when later passes read what an earlier one left', async () => { + await seedHistory(); + await createTask().task.run(signal); + expect(await remainingVersionIds()).toEqual(expectedVersionIds); + + const loggers = await runOverlappingPasses(); + + expect(await remainingVersionIds()).toEqual(expectedVersionIds); + for (const logger of loggers) { + expect(logger.error).not.toHaveBeenCalled(); + } + }); +}); diff --git a/packages/workflow/src/workflow-diff.ts b/packages/workflow/src/workflow-diff.ts index a0990fcf462..6c6e32077f2 100644 --- a/packages/workflow/src/workflow-diff.ts +++ b/packages/workflow/src/workflow-diff.ts @@ -286,7 +286,9 @@ function makeMergeDependingOnSizeRule(mapping: Map count) return time(prev, next); } - return false; + // The smallest threshold is inclusive, so a version with no nodes scores zero and still merges. + const smallest = pairs.at(-1); + return smallest?.[0] === metaData.workflowSizeScore ? smallest[1](prev, next) : false; }; } @@ -361,13 +363,10 @@ export function groupWorkflows( const n = remaining.length; const metaData = { - // check latest and an "average" workflow to get a somewhat accurate representation - // without counting through the entire history + // The newest version is never removed, so a pass over the survivors of a + // pass scores the same and removes nothing more. workflowSizeScore: metaDataFields?.workflowSizeScore - ? Math.max( - determineNodeParametersSize(workflows[Math.floor(workflows.length / 2)]), - determineNodeParametersSize(workflows[workflows.length - 1]), - ) + ? determineNodeParametersSize(workflows[workflows.length - 1]) : undefined, } satisfies DiffMetaData; diff --git a/packages/workflow/test/workflow-diff.test.ts b/packages/workflow/test/workflow-diff.test.ts index 82a65a57917..0d3e059059b 100644 --- a/packages/workflow/test/workflow-diff.test.ts +++ b/packages/workflow/test/workflow-diff.test.ts @@ -24,6 +24,7 @@ import { type DiffableWorkflow, type DiffMetaData, type DiffRule, + type WorkflowDiffBase, } from '../src/workflow-diff'; describe('NodeDiffStatus', () => { @@ -911,6 +912,19 @@ describe('groupWorkflows', () => { expect(rule(prev, next, wcs, metaData)).toBe(false); }); + it('should apply the smallest threshold to a workflow size of zero when that threshold is zero', () => { + const mapping = new Map([ + [0, 60000], + [1000, 600000], + ]); + const rule = RULES.makeMergeDependingOnSizeRule(mapping); + + const prev = createWorkflow(new Date('2024-01-01T10:00:00Z')); + const next = createWorkflow(new Date('2024-01-01T10:00:30Z')); + + expect(rule(prev, next, wcs, createMetaData(0))).toBe(true); + }); + it('should apply the correct time threshold for workflow size', () => { const mapping = new Map([ [1000, 60000], // 1000 chars -> 1 min @@ -1390,3 +1404,109 @@ describe('hasCredentialChanges', () => { expect(result).toBe(false); }); }); + +describe('groupWorkflows with the trim rule', () => { + const MINUTE_MS = 60 * 1_000; + + // The rule exactly as the workflow history trim task configures it. + const trimRule = RULES.makeMergeDependingOnSizeRule( + new Map([ + [0, MINUTE_MS], + [100, 10 * MINUTE_MS], + [1000, 120 * MINUTE_MS], + [5000, 300 * MINUTE_MS], + [10000, 600 * MINUTE_MS], + ]), + ); + + type Version = WorkflowDiffBase & { versionId: string; authors: string }; + type VersionOptions = { size?: 'small' | 'large'; authors?: string }; + + const version = ( + versionId: string, + seconds: number, + { size = 'small', authors = 'Test User' }: VersionOptions = {}, + ) => + mock({ + versionId, + createdAt: new Date(seconds * 1_000), + nodes: [{ id: 'n', name: 'n', parameters: { a: size === 'large' ? 'x'.repeat(200) : 'x' } }], + connections: {}, + name: null, + authors, + }); + + const trim = (versions: Version[]) => + groupWorkflows(versions, [trimRule], [SKIP_RULES.skipDifferentUsers], { + workflowSizeScore: true, + }).remaining.map((v) => v.versionId); + + const survivorsOf = (versions: Version[]) => { + const kept = trim(versions); + return versions.filter((v) => kept.includes(v.versionId)); + }; + + it('keeps a version whose gap to the kept neighbour equals the bucket, merges one a millisecond short', () => { + expect(trim([version('v0', 0), version('v1', 60)])).toEqual(['v0', 'v1']); + expect(trim([version('v0', 0.001), version('v1', 60)])).toEqual(['v1']); + }); + + it('measures the gap to the kept neighbour, not to a merged one', () => { + expect(trim([version('v0', 0), version('v1', 50), version('v2', 100)])).toEqual(['v0', 'v2']); + }); + + it('takes the size score from the newest version', () => { + // The middle version is large. The newest is small, so the bucket is one minute. + const versions = [ + version('v0', 0, { size: 'large' }), + version('v1', 300, { size: 'large' }), + version('v2', 600), + ]; + + expect(trim(versions)).toEqual(['v0', 'v1', 'v2']); + }); + + it('removes nothing from the survivors of a pass, mixed sizes', () => { + const versions = [ + version('v0', 0), + version('v1', 100), + version('v2', 200, { size: 'large' }), + version('v3', 280), + version('v4', 300), + version('v5', 380), + version('v6', 400), + ]; + const once = trim(versions); + + expect(once).toEqual(['v0', 'v1', 'v2', 'v4', 'v6']); + expect(trim(survivorsOf(versions))).toEqual(once); + }); + + it('widens the bucket when the newest version is large', () => { + // A score above 100 keeps one version per ten minutes. + const versions = [version('v0', 1), version('v1', 300), version('v2', 600, { size: 'large' })]; + + expect(trim(versions)).toEqual(['v2']); + }); + + it('trims a workflow whose newest version has no nodes', () => { + const versions = [version('v0', 0, { size: 'large' }), version('v1', 30), version('v2', 60)]; + versions[2].nodes = []; + + expect(trim(versions)).toEqual(['v0', 'v2']); + }); + + it('removes nothing from the survivors of a pass, two authors', () => { + const versions = [ + version('v0', 0), + version('v1', 20), + version('v2', 40, { authors: 'other' }), + version('v3', 50, { authors: 'other' }), + version('v4', 70), + ]; + const once = trim(versions); + + expect(once).toEqual(['v1', 'v3', 'v4']); + expect(trim(survivorsOf(versions))).toEqual(once); + }); +});