github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 05:29:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

feat(core): Run the workflow history trim as a durable daily system task (#40085)

Browse Source
This commit is contained in:
Veljko Terzic 2026-10-06 12:17:32 +00:00 committed by GitHub
parent c4112794fa
commit 38eea65993
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 722 additions and 247 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
packages/@n8n/db/src/repositories/workflow-history.repository.ts
View File

@ -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<WorkflowHistory> {
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<WorkflowHistory> {
skipRules: DiffRule[] = [],
metaData?: Partial<Record<keyof DiffMetaData, boolean>>,
): 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<WorkflowHistory> {
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<WorkflowHistory>(
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 };
}

11
packages/@n8n/db/src/repositories/workflow-publish-history.repository.ts
View File

@ -38,17 +38,6 @@ export class WorkflowPublishHistoryRepository extends Repository<WorkflowPublish
return await repository.find({ where: { workflowId, versionId }, order: { id: 'ASC' } });
}
async getPublishedVersions(
workflowId: string,
): Promise<Array<Pick<WorkflowPublishHistory, 'versionId'>>> {
return await this.manager
.createQueryBuilder(WorkflowPublishHistory, 'wph')
.select('wph.versionId')
.distinct(true)
.where('wph.workflowId = :workflowId', { workflowId })
.getMany();
}
async findActivatedByUserId(workflowId: string): Promise<string | undefined> {
const record = await this.findOne({
select: ['userId'],

12
packages/cli/src/scheduling/system-tasks/__tests__/main-system-tasks.test.ts
View File

@ -18,12 +18,15 @@ const configWith = ({
pruneData = true,
useWorkflowPublicationService = true,
diagnosticsEnabled = true,
historyPruneTimeHours = -1,
dbType = 'postgresdb',
} = {}) =>
mock<GlobalConfig>({
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 }));

19
packages/cli/src/scheduling/system-tasks/__tests__/system-task-runner.test.ts
View File

@ -34,6 +34,7 @@ describe('SystemTaskRunner', () => {
enabledForSystemTasks = false,
instanceRole = 'leader' as InstanceSettings['instanceRole'],
instanceType = 'main' as InstanceSettings['instanceType'],
timezone = 'UTC',
} = {}) {
const logger = mock<Logger>();
const metadata = new SystemTaskMetadata();
@ -54,7 +55,7 @@ describe('SystemTaskRunner', () => {
jobRegistrar,
systemTaskOwner,
mock<GlobalConfig>({
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);

12
packages/cli/src/scheduling/system-tasks/main-system-tasks.ts
View File

@ -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<Syste
const { WorkflowHistoryCompactionOptimizeTask } = await import(
'@/services/pruning/workflow-history-compaction-optimize.task.js'
);
const { WorkflowHistoryCompactionTrimTask } = await import(
'@/services/pruning/workflow-history-compaction-trim.task.js'
);
const { WorkflowHistoryPruningTask } = await import(
'@/services/pruning/workflow-history-pruning.task.js'
);
@ -26,11 +24,17 @@ export async function mainSystemTasks(globalConfig: GlobalConfig): Promise<Syste
ActivityPruningTask,
LicenseRenewalTask,
WorkflowHistoryCompactionOptimizeTask,
WorkflowHistoryCompactionTrimTask,
WorkflowHistoryPruningTask,
PendingAuthorizationCleanupTask,
];
if (isTrimmingEnabled(globalConfig.workflowHistory, globalConfig.workflowHistoryCompaction)) {
const { WorkflowHistoryCompactionTrimTask } = await import(
'@/services/pruning/workflow-history-compaction-trim.task.js'
);
tasks.push(WorkflowHistoryCompactionTrimTask);
}
if (globalConfig.executions.pruneData) {
const { ExecutionPruningSoftDeleteTask } = await import(
'@/services/pruning/execution-pruning-soft-delete.task.js'

50
packages/cli/src/services/pruning/__tests__/workflow-history-compaction-trim.task.test.ts
View File

@ -1,57 +1,29 @@
import { Time } from '@n8n/constants';
import { mock } from 'vitest-mock-extended';
import { WorkflowHistoryCompactionTrimTask } from '../workflow-history-compaction-trim.task';
import type { WorkflowHistoryCompactionService } from '../workflow-history-compaction.service';
describe('WorkflowHistoryCompactionTrimTask', () => {
let compactionService = mock<WorkflowHistoryCompactionService>();
let task = new WorkflowHistoryCompactionTrimTask(compactionService);
const compactionService = mock<WorkflowHistoryCompactionService>();
const task = new WorkflowHistoryCompactionTrimTask(compactionService);
const setService = ({ trimmingEnabled = true } = {}) => {
Object.defineProperty(compactionService, 'isTrimmingEnabled', { value: trimmingEnabled });
};
beforeEach(() => {
compactionService = mock<WorkflowHistoryCompactionService>();
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();
});
});

168
packages/cli/src/services/pruning/__tests__/workflow-history-compaction.service.test.ts
View File

@ -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<DbConnection>({
@ -26,37 +23,21 @@ describe('WorkflowHistoryCompactionService', () => {
skipOnStartUp: false,
});
const globalConfig = mock<GlobalConfig>({
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<InstanceSettings>({ 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<InstanceSettings>({ isLeader: true, instanceType: 'main', isMultiMain: true }),
dbConnection,
mock(),
mock<EventService>(),
);
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>();
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>();
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>();
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>();
workflowHistoryRepository.getWorkflowIdsInRange.mockResolvedValue([]);
const compactingService = new WorkflowHistoryCompactionService(
config,
mock<GlobalConfig>({ generic: { timezone }, workflowHistory: { pruneTime: -1 } }),
mockLogger(),
mock<InstanceSettings>({ isLeader: true, instanceType: 'main', isMultiMain: true }),
dbConnection,
workflowHistoryRepository,
mock<EventService>(),
);
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>();
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>();
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<InstanceSettings>({ isLeader: true, instanceType: 'main', isMultiMain: true }),
dbConnection,
mock(),
mock<EventService>(),
);
vi.spyOn(compactingService, 'optimizeHistories').mockImplementation((() => {}) as never);
const trimLongRunningHistoriesSpy = vi
.spyOn(compactingService, 'trimLongRunningHistories')
.mockImplementation((() => {}) as never);
compactingService.runStartupCompaction();
expect(trimLongRunningHistoriesSpy).not.toHaveBeenCalled();
vi.useRealTimers();
});
});

19
packages/cli/src/services/pruning/__tests__/workflow-history-compaction.utils.test.ts Normal file
View File

@ -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,
});
});
});

19
packages/cli/src/services/pruning/workflow-history-compaction-trim.task.ts
View File

@ -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<void> {
if (!this.compactionService.isTrimmingEnabled) return;
if (new Date().getHours() !== 3) return;
await this.compactionService.trimLongRunningHistories(signal);
}
}

33
packages/cli/src/services/pruning/workflow-history-compaction.service.ts
View File

@ -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<void> {
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();

18
packages/cli/src/services/pruning/workflow-history-compaction.utils.ts Normal file
View File

@ -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
);
}

290
packages/cli/test/integration/database/repositories/workflow-history.repository.test.ts
View File

@ -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<typeof originalDelete>[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<typeof originalDelete>[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

137
packages/cli/test/integration/workflow-history-compaction-trim.task.test.ts Normal file
View File

@ -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<InstanceSettings>({ instanceType: 'main', isLeader: false }),
Container.get(DbConnection),
repository,
mock<EventService>(),
);
return { task: new WorkflowHistoryCompactionTrimTask(service), logger };
}
async function seedHistory(): Promise<void> {
// 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<string[]> {
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<Logger[]> {
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();
}
});
});

13
packages/workflow/src/workflow-diff.ts
View File

@ -286,7 +286,9 @@ function makeMergeDependingOnSizeRule<W extends DiffableWorkflow>(mapping: Map<n
for (const [count, time] of pairs) {
if (metaData.workflowSizeScore > 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<W extends WorkflowDiffBase = WorkflowDiffBase>(
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;

120
packages/workflow/test/workflow-diff.test.ts
View File

@ -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<Version>({
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);
});
});