github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 21:49:36 +02:00
Code Issues Actions 13 Packages Projects Releases Wiki Activity Graph

refactor(core): Run insights compaction and pruning as system tasks (no-changelog) (#37543)

Browse Source 
Co-authored-by: Veljko Terzic <veljko.terzic@n8n.io>
This commit is contained in:
Lorent Lempereur 2026-09-08 07:15:14 +00:00 committed by GitHub
parent cbaa0be44d
commit 52befceb66
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 415 additions and 499 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
packages/cli/eslint.config.mjs
View File

@ -418,7 +418,6 @@ export default defineConfig(
files: [
'./src/license.ts',
'./src/modules/agents/integrations/n8n-checkpoint-storage.ts',
'./src/modules/insights/insights.service.ts',
'./src/modules/instance-ai/instance-ai.service.ts',
'./src/modules/instance-reporting/instance-reporting-scheduler.service.ts',
'./src/services/pruning/executions-pruning.service.ts',

305
packages/cli/src/modules/insights/__tests__/insights-compaction.service.integration.test.ts
View File

@ -1,11 +1,5 @@
import type { Logger } from '@n8n/backend-common';
import {
mockLogger,
createTeamProject,
createWorkflow,
testDb,
testModules,
} from '@n8n/backend-test-utils';
import { createTeamProject, createWorkflow, testDb, testModules } from '@n8n/backend-test-utils';
import type { WorkflowEntity } from '@n8n/db';
import { Container } from '@n8n/di';
import { DateTime } from 'luxon';
@ -181,8 +175,10 @@ describe('compaction', () => {
timestamp = timestamp.plus({ minute: 1 });
}
const controller = new AbortController();
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expect(insightsRawRepository.count()).resolves.toBe(0);
@ -227,8 +223,10 @@ describe('compaction', () => {
timestamp = timestamp.plus({ minute: 1 });
}
const controller = new AbortController();
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expect(insightsRawRepository.count()).resolves.toBe(0);
@ -323,8 +321,10 @@ describe('compaction', () => {
}
await createRawInsightsEvents(workflow, events);
const controller = new AbortController();
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expect(insightsRawRepository.count()).resolves.toBe(0);
@ -356,8 +356,10 @@ describe('compaction', () => {
}
await createRawInsightsEvents(workflow, events);
const controller = new AbortController();
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expect(insightsRawRepository.count()).resolves.toBe(0);
@ -367,171 +369,7 @@ describe('compaction', () => {
});
});
describe('compactionSchedule', () => {
test('compaction is running on schedule', async () => {
// ARRANGE
vi.useFakeTimers();
const insightsCompactionService = new InsightsCompactionService(
mock<InsightsByPeriodRepository>(),
mock<InsightsRawRepository>(),
mock<InsightsConfig>({
compactionIntervalMinutes: 60,
}),
mockLogger(),
);
// spy on the compactInsights method to check if it's called
const compactInsightsSpy = vi
.spyOn(insightsCompactionService, 'compactInsights')
.mockResolvedValue();
try {
insightsCompactionService.startCompactionTimer();
// ACT
// advance by 1 hour and 1 minute
vi.advanceTimersByTime(1000 * 60 * 61);
// ASSERT
expect(compactInsightsSpy).toHaveBeenCalledTimes(1);
} finally {
await insightsCompactionService.stopCompactionTimer();
vi.useRealTimers();
}
});
test('stop waits for the active compaction after a skipped overlapping interval', async () => {
// ARRANGE
vi.useFakeTimers();
const loggerDebug = vi.fn();
const logger = mock<Logger>({ scoped: vi.fn().mockReturnThis(), debug: loggerDebug });
const insightsCompactionService = new InsightsCompactionService(
mock<InsightsByPeriodRepository>(),
mock<InsightsRawRepository>(),
mock<InsightsConfig>({
compactionIntervalMinutes: 1,
compactionBatchSize: 2,
compactionBatchDelayMilliseconds: 0,
compactionMaxBatchesPerRun: 0,
compactionMaxRuntimeSeconds: 0,
}),
logger,
);
let resolveRawToHour!: (numberOfCompactedRawData: number) => void;
const firstRawToHourPromise = new Promise<number>((resolve) => {
resolveRawToHour = resolve;
});
const rawToHourSpy = vi
.spyOn(insightsCompactionService, 'compactRawToHour')
.mockReturnValueOnce(firstRawToHourPromise)
.mockResolvedValue(0);
vi.spyOn(insightsCompactionService, 'compactHourToDay').mockResolvedValue(0);
vi.spyOn(insightsCompactionService, 'compactDayToWeek').mockResolvedValue(0);
try {
insightsCompactionService.startCompactionTimer();
// ACT
vi.advanceTimersByTime(1000 * 60);
await Promise.resolve();
vi.advanceTimersByTime(1000 * 60);
await Promise.resolve();
const stopPromise = insightsCompactionService.stopCompactionTimer();
let stopped = false;
void stopPromise.then(() => {
stopped = true;
});
await Promise.resolve();
// ASSERT
expect(stopped).toBe(false);
expect(loggerDebug).toHaveBeenCalledWith(
'Skipping insights compaction because another compaction run is active',
);
resolveRawToHour(0);
await stopPromise;
expect(stopped).toBe(true);
expect(rawToHourSpy).toHaveBeenCalledTimes(1);
} finally {
await insightsCompactionService.stopCompactionTimer();
vi.useRealTimers();
}
});
test('stop succeeds after compaction fails', async () => {
// ARRANGE
const insightsCompactionService = new InsightsCompactionService(
mock<InsightsByPeriodRepository>(),
mock<InsightsRawRepository>(),
mock<InsightsConfig>({
compactionBatchSize: 2,
compactionBatchDelayMilliseconds: 0,
compactionMaxBatchesPerRun: 0,
compactionMaxRuntimeSeconds: 0,
}),
mockLogger(),
);
vi.spyOn(insightsCompactionService, 'compactRawToHour').mockRejectedValueOnce(
new Error('compaction failed'),
);
// ACT
const compactionPromise = insightsCompactionService.compactInsights();
const stopPromise = insightsCompactionService.stopCompactionTimer();
// ASSERT
await expect(compactionPromise).rejects.toThrow('compaction failed');
await expect(stopPromise).resolves.toBeUndefined();
});
});
describe('compactInsights in-memory run guard', () => {
test('skips a concurrent run and accepts another run after the active one finishes', async () => {
// ARRANGE
const logger = mock<Logger>({ scoped: vi.fn().mockReturnThis() });
const insightsCompactionService = new InsightsCompactionService(
mock<InsightsByPeriodRepository>(),
mock<InsightsRawRepository>(),
mock<InsightsConfig>({
compactionBatchSize: 2,
compactionBatchDelayMilliseconds: 0,
compactionMaxBatchesPerRun: 0,
compactionMaxRuntimeSeconds: 0,
}),
logger,
);
let resolveRawToHour!: (numberOfCompactedRawData: number) => void;
const firstRawToHourPromise = new Promise<number>((resolve) => {
resolveRawToHour = resolve;
});
const rawToHourSpy = vi
.spyOn(insightsCompactionService, 'compactRawToHour')
.mockReturnValueOnce(firstRawToHourPromise)
.mockResolvedValue(0);
vi.spyOn(insightsCompactionService, 'compactHourToDay').mockResolvedValue(0);
vi.spyOn(insightsCompactionService, 'compactDayToWeek').mockResolvedValue(0);
// ACT
const firstCompactionPromise = insightsCompactionService.compactInsights();
await Promise.resolve();
await insightsCompactionService.compactInsights();
// ASSERT
expect(logger.debug).toHaveBeenCalledWith(
'Skipping insights compaction because another compaction run is active',
);
resolveRawToHour(0);
await firstCompactionPromise;
await insightsCompactionService.compactInsights();
expect(rawToHourSpy).toHaveBeenCalledTimes(2);
});
describe('compactInsights failure recovery', () => {
test('accepts another run after the active run fails', async () => {
// ARRANGE
const logger = mock<Logger>({ scoped: vi.fn().mockReturnThis() });
@ -557,12 +395,14 @@ describe('compaction', () => {
.spyOn(insightsCompactionService, 'compactDayToWeek')
.mockResolvedValue(0);
const controller = new AbortController();
// ACT + ASSERT
await expect(insightsCompactionService.compactInsights()).rejects.toThrow(
await expect(insightsCompactionService.compactInsights(controller.signal)).rejects.toThrow(
'compaction failed',
);
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
expect(rawToHourSpy).toHaveBeenCalledTimes(2);
expect(hourToDaySpy).toHaveBeenCalledTimes(1);
expect(dayToWeekSpy).toHaveBeenCalledTimes(1);
@ -583,15 +423,17 @@ describe('compaction', () => {
const workflow = await createWorkflow({}, project);
await createRawSuccessEvents(workflow, 5);
const controller = new AbortController();
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expectRawCount(1);
await expect(getCompactedTotal('hour')).resolves.toBe(4);
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expectRawCount(0);
@ -637,8 +479,10 @@ describe('compaction', () => {
.startOf('day'),
});
const controller = new AbortController();
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expectRawCount(0);
@ -665,8 +509,10 @@ describe('compaction', () => {
.mockReturnValueOnce(0)
.mockReturnValue(1000);
const controller = new AbortController();
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expectRawCount(3);
@ -676,7 +522,7 @@ describe('compaction', () => {
overrideCompactionConfig({ compactionMaxRuntimeSeconds: 0 });
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expectRawCount(0);
@ -684,6 +530,93 @@ describe('compaction', () => {
});
});
describe('compactInsights abort', () => {
test('compacts nothing when the signal is already aborted', async () => {
// ARRANGE
overrideCompactionConfig({
compactionBatchSize: 2,
compactionMaxBatchesPerRun: 0,
compactionMaxRuntimeSeconds: 0,
compactionBatchDelayMilliseconds: 0,
});
const insightsCompactionService = Container.get(InsightsCompactionService);
const project = await createTeamProject();
const workflow = await createWorkflow({}, project);
await createRawSuccessEvents(workflow, 5);
const controller = new AbortController();
controller.abort();
// ACT
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expectRawCount(5);
await expect(getCompactedTotal('hour')).resolves.toBe(0);
});
test('stops before the next batch once aborted and resumes on a later run', async () => {
// ARRANGE
overrideCompactionConfig({
compactionBatchSize: 2,
compactionMaxBatchesPerRun: 0,
compactionMaxRuntimeSeconds: 0,
compactionBatchDelayMilliseconds: 0,
});
const insightsCompactionService = Container.get(InsightsCompactionService);
const project = await createTeamProject();
const workflow = await createWorkflow({}, project);
await createRawSuccessEvents(workflow, 5);
const controller = new AbortController();
const compactRawToHour =
insightsCompactionService.compactRawToHour.bind(insightsCompactionService);
vi.spyOn(insightsCompactionService, 'compactRawToHour').mockImplementationOnce(async () => {
const compacted = await compactRawToHour();
controller.abort();
return compacted;
});
// ACT
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expectRawCount(3);
await expect(getCompactedTotal('hour')).resolves.toBe(2);
// ACT
const laterRunController = new AbortController();
await insightsCompactionService.compactInsights(laterRunController.signal);
// ASSERT
await expectRawCount(0);
await expect(getCompactedTotal('hour')).resolves.toBe(5);
});
test('ends the inter-batch delay early when aborted during it', async () => {
// ARRANGE
overrideCompactionConfig({
compactionBatchSize: 2,
compactionMaxBatchesPerRun: 0,
compactionMaxRuntimeSeconds: 0,
compactionBatchDelayMilliseconds: 10_000,
});
const insightsCompactionService = Container.get(InsightsCompactionService);
const project = await createTeamProject();
const workflow = await createWorkflow({}, project);
await createRawSuccessEvents(workflow, 5);
const controller = new AbortController();
setTimeout(() => controller.abort(), 50);
const startedAt = Date.now();
// ACT
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
expect(Date.now() - startedAt).toBeLessThan(5_000);
await expectRawCount(3);
await expect(getCompactedTotal('hour')).resolves.toBe(2);
});
});
describe('raw compaction batch ordering', () => {
test('processes rows with identical timestamps in id order across limited runs', async () => {
// ARRANGE
@ -706,8 +639,10 @@ describe('compaction', () => {
);
}
const controller = new AbortController();
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
const remainingAfterFirstRun = await insightsRawRepository.find({ order: { id: 'ASC' } });
@ -717,7 +652,7 @@ describe('compaction', () => {
await expect(getCompactedTotal('hour')).resolves.toBe(2);
// ACT
await insightsCompactionService.compactInsights();
await insightsCompactionService.compactInsights(controller.signal);
// ASSERT
await expectRawCount(0);

29
packages/cli/src/modules/insights/__tests__/insights-compaction.task.test.ts Normal file
View File

@ -0,0 +1,29 @@
import { mock } from 'vitest-mock-extended';
import type { InsightsCompactionService } from '../insights-compaction.service';
import { InsightsCompactionTask } from '../insights-compaction.task';
import { InsightsConfig } from '../insights.config';
describe('InsightsCompactionTask', () => {
const insightsConfig = new InsightsConfig();
const compactionService = mock<InsightsCompactionService>();
const task = new InsightsCompactionTask(insightsConfig, compactionService);
it('should declare the configured compaction cadence', () => {
expect(task.name).toBe('insights-compaction');
expect(task.schedule).toEqual({
kind: 'interval',
intervalSeconds: insightsConfig.compactionIntervalMinutes * 60,
});
expect(task.effects).toBe('idempotent');
expect(task.durable).toBe(false);
});
it('should compact insights on run, handing it the run signal', async () => {
const { signal } = new AbortController();
await task.run(signal);
expect(compactionService.compactInsights).toHaveBeenCalledExactlyOnceWith(signal);
});
});

103
packages/cli/src/modules/insights/__tests__/insights-pruning.service.test.ts
View File

@ -1,20 +1,19 @@
import { mockLogger } from '@n8n/backend-test-utils';
import { Time } from '@n8n/constants';
import { mock } from 'vitest-mock-extended';
import type { InsightsByPeriodRepository } from '../database/repositories/insights-by-period.repository';
import { InsightsPruningService } from '../insights-pruning.service';
import { InsightsConfig } from '../insights.config';
import { INSIGHTS_MAX_AGE_DAYS_CAP, INSIGHTS_MAX_AGE_DAYS_DEFAULT } from '../insights.constants';
describe('InsightsPruningService', () => {
let insightsConfig: InsightsConfig;
let insightsByPeriodRepository: InsightsByPeriodRepository;
let insightsPruningService: InsightsPruningService;
beforeAll(() => {
beforeEach(() => {
insightsConfig = new InsightsConfig();
insightsConfig.maxAgeDays = 10;
insightsConfig.pruneCheckIntervalHours = 1;
insightsByPeriodRepository = mock<InsightsByPeriodRepository>();
insightsPruningService = new InsightsPruningService(
insightsByPeriodRepository,
@ -23,90 +22,42 @@ describe('InsightsPruningService', () => {
);
});
describe('pruning scheduling', () => {
beforeEach(() => {
vi.useFakeTimers();
insightsPruningService.startPruningTimer();
describe('pruningMaxAgeInDays', () => {
test('returns the configured value', () => {
expect(insightsPruningService.pruningMaxAgeInDays).toBe(10);
});
afterEach(() => {
vi.useRealTimers();
insightsPruningService.stopPruningTimer();
vi.restoreAllMocks();
vi.clearAllMocks();
test('caps the configured value', () => {
insightsConfig.maxAgeDays = INSIGHTS_MAX_AGE_DAYS_CAP + 1;
expect(insightsPruningService.pruningMaxAgeInDays).toBe(INSIGHTS_MAX_AGE_DAYS_CAP);
});
test('pruning timeout is scheduled on start and rescheduled after each run', async () => {
const insightsByPeriodRepository = mock<InsightsByPeriodRepository>({
pruneOldData: async () => {
return { affected: 0 };
},
});
const insightsPruningService = new InsightsPruningService(
insightsByPeriodRepository,
insightsConfig,
mockLogger(),
);
const pruneSpy = vi.spyOn(insightsPruningService, 'pruneInsights');
const scheduleNextPruneSpy = vi.spyOn(insightsPruningService as any, 'scheduleNextPrune');
insightsPruningService.startPruningTimer();
// Wait for pruning timer promise to resolve
await vi.advanceTimersToNextTimerAsync();
expect(pruneSpy).toHaveBeenCalledTimes(1);
expect(scheduleNextPruneSpy).toHaveBeenCalledTimes(2);
test('maps -1 (keep forever) to the cap', () => {
insightsConfig.maxAgeDays = -1;
expect(insightsPruningService.pruningMaxAgeInDays).toBe(INSIGHTS_MAX_AGE_DAYS_CAP);
});
test('if stopped during prune, it does not reschedule the timeout', async () => {
const insightsByPeriodRepository = mock<InsightsByPeriodRepository>({
pruneOldData: async () => {
return { affected: 0 };
},
});
const insightsPruningService = new InsightsPruningService(
insightsByPeriodRepository,
insightsConfig,
mockLogger(),
);
let resolvePrune!: () => void;
const pruneInsightsMock = vi
.spyOn(insightsPruningService, 'pruneInsights')
.mockImplementation(
async () =>
await new Promise((resolve) => {
resolvePrune = () => resolve();
}),
);
insightsConfig.pruneCheckIntervalHours = 1;
insightsPruningService.startPruningTimer();
vi.advanceTimersByTime(Time.hours.toMilliseconds + 1); // 1h + 1min
// Immediately stop while pruning is "in progress"
insightsPruningService.stopPruningTimer();
resolvePrune(); // Now allow the fake pruning to complete
// Wait for pruning timer promise and reschedule to resolve
await vi.runOnlyPendingTimersAsync();
expect(pruneInsightsMock).toHaveBeenCalledTimes(1); // Only from start, not re-scheduled
test('falls back to the default for invalid values', () => {
insightsConfig.maxAgeDays = 0;
expect(insightsPruningService.pruningMaxAgeInDays).toBe(INSIGHTS_MAX_AGE_DAYS_DEFAULT);
});
});
test('pruneInsights is retried up when failing', async () => {
const pruneOldDataSpy = vi
.mocked(insightsByPeriodRepository.pruneOldData)
.mockRejectedValueOnce(new Error('Fail 1'))
.mockRejectedValueOnce(new Error('Fail 2'))
.mockResolvedValueOnce({ affected: 0 });
describe('pruneInsights', () => {
test('deletes data past the retention window', async () => {
vi.mocked(insightsByPeriodRepository.pruneOldData).mockResolvedValueOnce({ affected: 42 });
await insightsPruningService.pruneInsights();
await vi.advanceTimersByTimeAsync(Time.seconds.toMilliseconds * 2 + 1);
expect(pruneOldDataSpy).toHaveBeenCalledTimes(3);
expect(insightsByPeriodRepository.pruneOldData).toHaveBeenCalledWith(10);
});
test('propagates a pruning failure to the caller', async () => {
vi.mocked(insightsByPeriodRepository.pruneOldData).mockRejectedValueOnce(
new Error('prune failed'),
);
await expect(insightsPruningService.pruneInsights()).rejects.toThrow('prune failed');
});
});
});

28
packages/cli/src/modules/insights/__tests__/insights-pruning.task.test.ts Normal file
View File

@ -0,0 +1,28 @@
import { mock } from 'vitest-mock-extended';
import type { InsightsPruningService } from '../insights-pruning.service';
import { InsightsPruningTask } from '../insights-pruning.task';
import { InsightsConfig } from '../insights.config';
describe('InsightsPruningTask', () => {
const insightsConfig = new InsightsConfig();
const pruningService = mock<InsightsPruningService>();
const task = new InsightsPruningTask(insightsConfig, pruningService);
it('should declare the configured prune-check cadence', () => {
expect(task.name).toBe('insights-pruning');
expect(task.schedule).toEqual({
kind: 'interval',
intervalSeconds: insightsConfig.pruneCheckIntervalHours * 3600,
});
expect(task.effects).toBe('idempotent');
expect(task.durable).toBe(false);
expect(task.retryDelaySeconds).toBe(1);
});
it('should prune insights on run', async () => {
await task.run();
expect(pruningService.pruneInsights).toHaveBeenCalledTimes(1);
});
});

42
packages/cli/src/modules/insights/__tests__/insights.config.test.ts Normal file
View File

@ -0,0 +1,42 @@
import { Container } from '@n8n/di';
import { InsightsConfig } from '../insights.config';
describe('InsightsConfig', () => {
beforeEach(() => {
Container.reset();
});
afterEach(() => {
delete process.env.N8N_INSIGHTS_PRUNE_CHECK_INTERVAL_HOURS;
delete process.env.N8N_INSIGHTS_COMPACTION_INTERVAL_MINUTES;
});
it('reads the cadences from their environment variables', () => {
process.env.N8N_INSIGHTS_PRUNE_CHECK_INTERVAL_HOURS = '6';
process.env.N8N_INSIGHTS_COMPACTION_INTERVAL_MINUTES = '0.5';
const config = Container.get(InsightsConfig);
expect(config.pruneCheckIntervalHours).toBe(6);
expect(config.compactionIntervalMinutes).toBe(0.5);
});
it.each(['0', '-1', 'abc', 'Infinity'])(
'falls back to the default prune-check cadence when given %s',
(value) => {
process.env.N8N_INSIGHTS_PRUNE_CHECK_INTERVAL_HOURS = value;
expect(Container.get(InsightsConfig).pruneCheckIntervalHours).toBe(24);
},
);
it.each(['0', '-1', 'abc', 'Infinity'])(
'falls back to the default compaction cadence when given %s',
(value) => {
process.env.N8N_INSIGHTS_COMPACTION_INTERVAL_MINUTES = value;
expect(Container.get(InsightsConfig).compactionIntervalMinutes).toBe(60);
},
);
});

13
packages/cli/src/modules/insights/__tests__/insights.module.test.ts
View File

@ -8,6 +8,8 @@ import { mock } from 'vitest-mock-extended';
import { InsightsByPeriodRepository } from '../database/repositories/insights-by-period.repository';
import { InsightsCollectionService } from '../insights-collection.service';
import { InsightsCompactionTask } from '../insights-compaction.task';
import { InsightsPruningTask } from '../insights-pruning.task';
import { InsightsModule } from '../insights.module';
import { InsightsService } from '../insights.service';
@ -41,8 +43,6 @@ describe('InsightsModule', () => {
Container.set(
InsightsService,
new InsightsService(
mock(),
mock(),
mock(),
Container.get(LicenseState),
mockInstanceSettings,
@ -54,6 +54,15 @@ describe('InsightsModule', () => {
await createTeamProject();
});
describe('systemTasks()', () => {
it('should register the compaction and pruning tasks', async () => {
await expect(insightsModule.systemTasks()).resolves.toEqual([
InsightsCompactionTask,
InsightsPruningTask,
]);
});
});
describe('Dynamic conditional import of InsightsCollectionService', () => {
it('should not import InsightsCollectionService for worker instances', async () => {
// ARRANGE

91
packages/cli/src/modules/insights/__tests__/insights.service.integration.test.ts
View File

@ -26,8 +26,6 @@ import { createCompactedInsightsEvent } from '../database/entities/__tests__/db-
import type { InsightsByPeriod } from '../database/entities/insights-by-period';
import type { InsightsByPeriodRepository } from '../database/repositories/insights-by-period.repository';
import { InsightsCollectionService } from '../insights-collection.service';
import type { InsightsCompactionService } from '../insights-compaction.service';
import type { InsightsPruningService } from '../insights-pruning.service';
import { InsightsService } from '../insights.service';
describe('InsightsService (Integration)', () => {
@ -54,25 +52,19 @@ describe('InsightsService (Integration)', () => {
await testDb.terminate();
});
describe('startTimers', () => {
describe('collection lifecycle', () => {
let insightsService: InsightsService;
let compactionService: InsightsCompactionService;
let pruningService: InsightsPruningService;
let instanceSettings: MockProxy<InstanceSettings>;
let realCollectionService: InsightsCollectionService;
let initSpy: MockInstance;
let shutdownSpy: MockInstance;
beforeEach(() => {
compactionService = mock<InsightsCompactionService>();
pruningService = mock<InsightsPruningService>();
instanceSettings = mock<InstanceSettings>({
instanceType: 'main',
});
insightsService = new InsightsService(
mock<InsightsByPeriodRepository>(),
compactionService,
pruningService,
mock<LicenseState>(),
instanceSettings,
mockLogger(),
@ -95,51 +87,32 @@ describe('InsightsService (Integration)', () => {
shutdownSpy.mockRestore();
});
const setupMocks = (instanceType: InstanceType, isLeader: boolean = false) => {
const setupMocks = (instanceType: InstanceType) => {
(instanceSettings as any).instanceType = instanceType;
Object.defineProperty(instanceSettings, 'isLeader', {
get: vi.fn(() => isLeader),
});
};
test('starts flushing timer for main instance', async () => {
setupMocks('main', false);
test('starts collection for main instance', async () => {
setupMocks('main');
await insightsService.init();
expect(initSpy).toHaveBeenCalled();
expect(compactionService.startCompactionTimer).not.toHaveBeenCalled();
expect(pruningService.startPruningTimer).not.toHaveBeenCalled();
});
test('starts compaction, flushing and pruning timers for main leader instances', async () => {
setupMocks('main', true);
test('starts collection for webhook instance', async () => {
setupMocks('webhook');
await insightsService.init();
expect(initSpy).toHaveBeenCalled();
expect(compactionService.startCompactionTimer).toHaveBeenCalled();
expect(pruningService.startPruningTimer).toHaveBeenCalled();
});
test('starts only collection flushing timer for webhook instance', async () => {
setupMocks('webhook', false);
await insightsService.init();
expect(initSpy).toHaveBeenCalled();
expect(compactionService.startCompactionTimer).not.toHaveBeenCalled();
expect(pruningService.startPruningTimer).not.toHaveBeenCalled();
});
test('do no start any timers for non-main instances', async () => {
setupMocks('worker', false);
test('does not start collection for non-main instances', async () => {
setupMocks('worker');
await insightsService.init();
expect(initSpy).not.toHaveBeenCalled();
expect(compactionService.startCompactionTimer).not.toHaveBeenCalled();
expect(pruningService.startPruningTimer).not.toHaveBeenCalled();
});
});
@ -1440,8 +1413,6 @@ describe('InsightsService (Integration)', () => {
licenseStateMock = mock<LicenseState>();
insightsService = new InsightsService(
mock<InsightsByPeriodRepository>(),
mock<InsightsCompactionService>(),
mock<InsightsPruningService>(),
licenseStateMock,
mock<InstanceSettings>(),
mockLogger(),
@ -1538,19 +1509,9 @@ describe('InsightsService (Integration)', () => {
describe('shutdown', () => {
let insightsService: InsightsService;
const mockCompactionService = mock<InsightsCompactionService>({
stopCompactionTimer: vi.fn(),
});
const mockPruningService = mock<InsightsPruningService>({
stopPruningTimer: vi.fn(),
});
beforeAll(() => {
insightsService = new InsightsService(
mock<InsightsByPeriodRepository>(),
mockCompactionService,
mockPruningService,
mock<LicenseState>(),
mock<InstanceSettings>({ instanceType: 'main' }),
mockLogger(),
@ -1558,14 +1519,7 @@ describe('InsightsService (Integration)', () => {
);
});
beforeEach(() => {
mockCompactionService.stopCompactionTimer.mockReset();
mockCompactionService.stopCompactionTimer.mockResolvedValue(undefined);
mockPruningService.stopPruningTimer.mockReset();
mockPruningService.stopPruningTimer.mockReturnValue(undefined);
});
test('shutdown stops timers and shuts down services', async () => {
test('shutdown stops the collection service', async () => {
// ARRANGE
// Get the real service from the container and spy on it
const realCollectionService = Container.get(InsightsCollectionService);
@ -1576,33 +1530,6 @@ describe('InsightsService (Integration)', () => {
// ASSERT
expect(shutdownSpy).toHaveBeenCalled();
expect(mockCompactionService.stopCompactionTimer).toHaveBeenCalled();
expect(mockPruningService.stopPruningTimer).toHaveBeenCalled();
});
test('stops pruning before waiting for compaction to finish', async () => {
// ARRANGE
const callOrder: string[] = [];
let resolveCompaction!: () => void;
mockCompactionService.stopCompactionTimer.mockImplementation(async () => {
callOrder.push('compaction');
await new Promise<void>((resolve) => {
resolveCompaction = resolve;
});
});
mockPruningService.stopPruningTimer.mockImplementation(() => {
callOrder.push('pruning');
});
// ACT
const stopPromise = insightsService.stopCompactionAndPruningTimers();
await Promise.resolve();
// ASSERT
expect(callOrder).toEqual(['pruning', 'compaction']);
resolveCompaction();
await stopPromise;
});
});
});

8
packages/cli/src/modules/insights/__tests__/insights.service.test.ts
View File

@ -10,8 +10,6 @@ import type { WorkflowSharingService } from '@/workflows/workflow-sharing.servic
import { TypeToNumber } from '../database/entities/insights-shared';
import type { InsightsByPeriodRepository } from '../database/repositories/insights-by-period.repository';
import type { InsightsCompactionService } from '../insights-compaction.service';
import type { InsightsPruningService } from '../insights-pruning.service';
import { InsightsService } from '../insights.service';
vi.mock('@/permissions.ee/check-access', () => ({
@ -24,8 +22,6 @@ describe('InsightsService', () => {
let insightsService: InsightsService;
let mockInsightsByPeriodRepository: MockProxy<InsightsByPeriodRepository>;
let mockCompactionService: MockProxy<InsightsCompactionService>;
let mockPruningService: MockProxy<InsightsPruningService>;
let mockLicenseState: MockProxy<LicenseState>;
let mockInstanceSettings: MockProxy<InstanceSettings>;
let mockWorkflowSharingService: MockProxy<WorkflowSharingService>;
@ -34,8 +30,6 @@ describe('InsightsService', () => {
vi.clearAllMocks();
mockInsightsByPeriodRepository = mock<InsightsByPeriodRepository>();
mockCompactionService = mock<InsightsCompactionService>();
mockPruningService = mock<InsightsPruningService>();
mockLicenseState = mock<LicenseState>();
mockInstanceSettings = mock<InstanceSettings>();
mockWorkflowSharingService = mock<WorkflowSharingService>();
@ -43,8 +37,6 @@ describe('InsightsService', () => {
insightsService = new InsightsService(
mockInsightsByPeriodRepository,
mockCompactionService,
mockPruningService,
mockLicenseState,
mockInstanceSettings,
mockLogger(),

12
packages/cli/src/modules/insights/__tests__/whole-seconds.test.ts Normal file
View File

@ -0,0 +1,12 @@
import { wholeSeconds } from '../whole-seconds';
describe('wholeSeconds', () => {
it.each([
[3600, 3600],
[0.1 * 3600, 360],
[90.4, 90],
[0.2, 1],
])('rounds %s seconds to %s', (seconds, expected) => {
expect(wholeSeconds(seconds)).toBe(expected);
});
});

83
packages/cli/src/modules/insights/insights-compaction.service.ts
View File

@ -11,25 +11,19 @@ type CompactionRunState = {
startedAt: number;
batchesProcessed: number;
rowsCompacted: number;
signal: AbortSignal;
};
type CompactionStopReason = 'max-batches' | 'max-runtime';
type CompactionStopReason = 'max-batches' | 'max-runtime' | 'aborted';
/**
* This service is responsible for compacting lower granularity insights data
* into higher granularity to control the size of the insights data.
* into higher granularity to control the size of the insights data. The
* periodic cadence, overlap protection, and graceful stop live on the
* `insights-compaction` system task's runner.
*/
@Service()
export class InsightsCompactionService {
private compactInsightsTimer: NodeJS.Timeout | undefined;
/** Tracks the in-flight compaction run so stopping the timer can await it. */
private compactionPromise: Promise<void> | undefined;
private get isCompactionRunning() {
return this.compactionPromise !== undefined;
}
constructor(
private readonly insightsByPeriodRepository: InsightsByPeriodRepository,
private readonly insightsRawRepository: InsightsRawRepository,
@ -39,48 +33,13 @@ export class InsightsCompactionService {
this.logger = this.logger.scoped('insights');
}
startCompactionTimer() {
void this.stopCompactionTimer();
this.compactInsightsTimer = setInterval(
async () => await this.compactInsights(),
this.insightsConfig.compactionIntervalMinutes * Time.minutes.toMilliseconds,
);
this.logger.debug('Started compaction timer');
}
async stopCompactionTimer() {
if (this.compactInsightsTimer !== undefined) {
clearInterval(this.compactInsightsTimer);
this.compactInsightsTimer = undefined;
this.logger.debug('Stopped compaction timer');
}
// Wait for an in-flight run to finish so its transaction isn't left running
// against a connection that may be closed right after stopping.
await this.compactionPromise?.catch(() => {});
}
async compactInsights() {
if (this.isCompactionRunning) {
this.logger.debug('Skipping insights compaction because another compaction run is active');
return;
}
const compactionPromise = this.runCompaction();
const trackedCompactionPromise = compactionPromise.finally(() => {
if (this.compactionPromise === trackedCompactionPromise) {
this.compactionPromise = undefined;
}
});
this.compactionPromise = trackedCompactionPromise;
await this.compactionPromise;
}
private async runCompaction() {
/** One full compaction run: raw→hour, hour→day, day→week, in bounded batches. */
async compactInsights(signal: AbortSignal) {
const runState: CompactionRunState = {
startedAt: Date.now(),
batchesProcessed: 0,
rowsCompacted: 0,
signal,
};
const stoppedAfterRawToHour = await this.compactStage({
@ -148,7 +107,7 @@ export class InsightsCompactionService {
return true;
}
await this.waitBeforeNextBatchIfFull(numberOfCompactedData);
await this.waitBeforeNextBatchIfFull(numberOfCompactedData, runState);
} while (numberOfCompactedData === this.insightsConfig.compactionBatchSize);
return false;
@ -157,6 +116,10 @@ export class InsightsCompactionService {
private getCompactionRunStopReason(
runState: CompactionRunState,
): CompactionStopReason | undefined {
if (runState.signal.aborted) {
return 'aborted';
}
if (
this.insightsConfig.compactionMaxBatchesPerRun > 0 &&
runState.batchesProcessed >= this.insightsConfig.compactionMaxBatchesPerRun
@ -180,17 +143,25 @@ export class InsightsCompactionService {
stageName: string,
runState: CompactionRunState,
) {
this.logger.warn('Stopping insights compaction because a per-run limit was reached', {
const details = {
reason,
stageName,
batchesProcessed: runState.batchesProcessed,
rowsCompacted: runState.rowsCompacted,
compactionMaxBatchesPerRun: this.insightsConfig.compactionMaxBatchesPerRun,
compactionMaxRuntimeSeconds: this.insightsConfig.compactionMaxRuntimeSeconds,
});
};
if (reason === 'aborted') {
this.logger.debug('Stopping insights compaction because the run was aborted', details);
} else {
this.logger.warn('Stopping insights compaction because a per-run limit was reached', details);
}
}
private async waitBeforeNextBatchIfFull(numberOfCompactedData: number) {
private async waitBeforeNextBatchIfFull(
numberOfCompactedData: number,
runState: CompactionRunState,
) {
if (
numberOfCompactedData !== this.insightsConfig.compactionBatchSize ||
this.insightsConfig.compactionBatchDelayMilliseconds <= 0
@ -198,7 +169,11 @@ export class InsightsCompactionService {
return;
}
await sleep(this.insightsConfig.compactionBatchDelayMilliseconds);
try {
await sleep(this.insightsConfig.compactionBatchDelayMilliseconds, runState.signal);
} catch {
// `sleep` rejects only on abort, which the loop checks for on its own.
}
}
/**

36
packages/cli/src/modules/insights/insights-compaction.task.ts Normal file
View File

@ -0,0 +1,36 @@
import { Time } from '@n8n/constants';
import { SystemTask } from '@n8n/decorators';
import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators';
import { InsightsCompactionService } from './insights-compaction.service';
import { InsightsConfig } from './insights.config';
import { wholeSeconds } from './whole-seconds';
/**
* Rolls raw insight events up into per-period summaries, so the insights
* dashboards stay fast and the stored data stays small.
*/
@SystemTask()
export class InsightsCompactionTask implements SystemTask {
readonly name = 'insights-compaction';
readonly schedule: SystemTaskSchedule = {
kind: 'interval',
intervalSeconds: wholeSeconds(
this.insightsConfig.compactionIntervalMinutes * Time.minutes.toSeconds,
),
};
readonly effects: SystemTaskEffects = 'idempotent';
readonly durable = false;
constructor(
private readonly insightsConfig: InsightsConfig,
private readonly compactionService: InsightsCompactionService,
) {}
async run(signal: AbortSignal): Promise<void> {
await this.compactionService.compactInsights(signal);
}
}

58
packages/cli/src/modules/insights/insights-pruning.service.ts
View File

@ -1,7 +1,5 @@
import { Logger } from '@n8n/backend-common';
import { Time } from '@n8n/constants';
import { Service } from '@n8n/di';
import { strict } from 'assert';
import { InsightsByPeriodRepository } from './database/repositories/insights-by-period.repository';
import { InsightsConfig } from './insights.config';
@ -9,12 +7,6 @@ import { INSIGHTS_MAX_AGE_DAYS_CAP, INSIGHTS_MAX_AGE_DAYS_DEFAULT } from './insi
@Service()
export class InsightsPruningService {
private pruneInsightsTimeout: NodeJS.Timeout | undefined;
private isStopped = true;
private readonly delayOnError = Time.seconds.toMilliseconds;
constructor(
private readonly insightsByPeriodRepository: InsightsByPeriodRepository,
private readonly config: InsightsConfig,
@ -40,51 +32,13 @@ export class InsightsPruningService {
return Math.min(configuredMaxAgeDays, INSIGHTS_MAX_AGE_DAYS_CAP);
}
startPruningTimer() {
strict(this.isStopped);
this.clearPruningTimer();
this.isStopped = false;
this.scheduleNextPrune();
this.logger.debug('Started pruning timer');
}
private clearPruningTimer() {
if (this.pruneInsightsTimeout !== undefined) {
clearTimeout(this.pruneInsightsTimeout);
this.pruneInsightsTimeout = undefined;
}
}
stopPruningTimer() {
this.isStopped = true;
this.clearPruningTimer();
this.logger.debug('Stopped pruning timer');
}
private scheduleNextPrune(
delayMs = this.config.pruneCheckIntervalHours * Time.hours.toMilliseconds,
) {
if (this.isStopped) return;
this.pruneInsightsTimeout = setTimeout(async () => {
await this.pruneInsights();
}, delayMs);
}
/** Deletes insights data past the retention window. A failure propagates to the caller. */
async pruneInsights() {
this.logger.info('Pruning old insights data');
try {
const result = await this.insightsByPeriodRepository.pruneOldData(this.pruningMaxAgeInDays);
this.logger.debug(
'Deleted insights by period',
result.affected ? { count: result.affected } : {},
);
this.scheduleNextPrune();
} catch (error: unknown) {
this.logger.warn('Pruning failed', { error });
// In case of failure, we retry the operation after a shorter time
this.scheduleNextPrune(this.delayOnError);
}
const result = await this.insightsByPeriodRepository.pruneOldData(this.pruningMaxAgeInDays);
this.logger.debug(
'Deleted insights by period',
result.affected ? { count: result.affected } : {},
);
}
}

38
packages/cli/src/modules/insights/insights-pruning.task.ts Normal file
View File

@ -0,0 +1,38 @@
import { Time } from '@n8n/constants';
import { SystemTask } from '@n8n/decorators';
import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators';
import { InsightsPruningService } from './insights-pruning.service';
import { InsightsConfig } from './insights.config';
import { wholeSeconds } from './whole-seconds';
/**
* Deletes insights statistics older than the configured retention period, so
* old data does not pile up forever.
*/
@SystemTask()
export class InsightsPruningTask implements SystemTask {
readonly name = 'insights-pruning';
readonly schedule: SystemTaskSchedule = {
kind: 'interval',
intervalSeconds: wholeSeconds(
this.insightsConfig.pruneCheckIntervalHours * Time.hours.toSeconds,
),
};
readonly effects: SystemTaskEffects = 'idempotent';
readonly durable = false;
readonly retryDelaySeconds = 1;
constructor(
private readonly insightsConfig: InsightsConfig,
private readonly pruningService: InsightsPruningService,
) {}
async run(): Promise<void> {
await this.pruningService.pruneInsights();
}
}

8
packages/cli/src/modules/insights/insights.config.ts
View File

@ -1,14 +1,18 @@
import { Config, Env } from '@n8n/config';
import { z } from 'zod';
import { INSIGHTS_MAX_AGE_DAYS_DEFAULT } from './insights.constants';
/** A task cadence, which the scheduler rejects unless it is above zero. */
const cadenceSchema = z.coerce.number().positive().finite();
@Config
export class InsightsConfig {
/**
* The interval in minutes at which the insights data should be compacted.
* Default: 60
*/
@Env('N8N_INSIGHTS_COMPACTION_INTERVAL_MINUTES')
@Env('N8N_INSIGHTS_COMPACTION_INTERVAL_MINUTES', cadenceSchema)
compactionIntervalMinutes: number = 60;
/**
@ -57,7 +61,7 @@ export class InsightsConfig {
* How often (hours) insights data will be checked for regular deletion.
* Default: 24
*/
@Env('N8N_INSIGHTS_PRUNE_CHECK_INTERVAL_HOURS')
@Env('N8N_INSIGHTS_PRUNE_CHECK_INTERVAL_HOURS', cadenceSchema)
pruneCheckIntervalHours: number = 24;
/**

6
packages/cli/src/modules/insights/insights.module.ts
View File

@ -15,6 +15,12 @@ export class InsightsModule implements ModuleInterface {
await Container.get(InsightsService).init();
}
async systemTasks() {
const { InsightsCompactionTask } = await import('./insights-compaction.task.js');
const { InsightsPruningTask } = await import('./insights-pruning.task.js');
return [InsightsCompactionTask, InsightsPruningTask];
}
async entities() {
const { InsightsByPeriod } = await import('./database/entities/insights-by-period.js');
const { InsightsMetadata } = await import('./database/entities/insights-metadata.js');

20
packages/cli/src/modules/insights/insights.service.ts
View File

@ -1,7 +1,6 @@
import { type InsightsSummary } from '@n8n/api-types';
import { LicenseState, Logger } from '@n8n/backend-common';
import type { User } from '@n8n/db';
import { OnLeaderStepdown, OnLeaderTakeover } from '@n8n/decorators';
import { Container, Service } from '@n8n/di';
import { DateTime } from 'luxon';
import { InstanceSettings } from 'n8n-core';
@ -15,15 +14,11 @@ import type { PeriodUnit, TypeUnit } from './database/entities/insights-shared';
import { NumberToType, TypeToNumber } from './database/entities/insights-shared';
import type { InsightsAccessFilter } from './database/repositories/insights-by-period.repository';
import { InsightsByPeriodRepository } from './database/repositories/insights-by-period.repository';
import { InsightsCompactionService } from './insights-compaction.service';
import { InsightsPruningService } from './insights-pruning.service';
@Service()
export class InsightsService {
constructor(
private readonly insightsByPeriodRepository: InsightsByPeriodRepository,
private readonly compactionService: InsightsCompactionService,
private readonly pruningService: InsightsPruningService,
private readonly licenseState: LicenseState,
private readonly instanceSettings: InstanceSettings,
private readonly logger: Logger,
@ -52,25 +47,10 @@ export class InsightsService {
async init() {
await this.toggleCollectionService(true);
if (this.instanceSettings.isLeader) this.startCompactionAndPruningTimers();
}
@OnLeaderTakeover()
startCompactionAndPruningTimers() {
this.compactionService.startCompactionTimer();
this.pruningService.startPruningTimer();
}
@OnLeaderStepdown()
async stopCompactionAndPruningTimers() {
this.pruningService.stopPruningTimer();
await this.compactionService.stopCompactionTimer();
}
async shutdown() {
await this.toggleCollectionService(false);
await this.stopCompactionAndPruningTimers();
}
/**

4
packages/cli/src/modules/insights/whole-seconds.ts Normal file
View File

@ -0,0 +1,4 @@
/** Rounds to the whole second the scheduler requires, never below one. */
export function wholeSeconds(seconds: number): number {
return Math.max(1, Math.round(seconds));
}

29
packages/cli/test/integration/insights/insights-vs-workflow-statistics.integration.test.ts
View File

@ -4,7 +4,7 @@
* properly distinguishing between root executions and subworkflow executions.
*
* This test actually executes workflows (not just mocking) to ensure end-to-end correctness.
* It configures the system for fast compaction and waits for automatic processing.
* It configures the system for fast flushing and drives compaction explicitly.
*/
import { createTeamProject, createWorkflow, testDb, testModules } from '@n8n/backend-test-utils';
@ -30,7 +30,6 @@ describe('Insights vs Workflow Statistics Integration', () => {
// Configure insights for fast flushing and compaction BEFORE loading modules
process.env.N8N_INSIGHTS_FLUSH_BATCH_SIZE = '10'; // Flush after 10 events
process.env.N8N_INSIGHTS_FLUSH_INTERVAL_SECONDS = '1'; // Flush every 1 second
process.env.N8N_INSIGHTS_COMPACTION_INTERVAL_MINUTES = '0.05'; // Compact every ~3 seconds
process.env.N8N_INSIGHTS_COMPACTION_BATCH_SIZE = '100'; // Process up to 100 items per batch
await testModules.loadModules(['insights']);
@ -94,15 +93,6 @@ describe('Insights vs Workflow Statistics Integration', () => {
// Initialize insights collection service (config already set via env vars)
insightsCollectionService.init();
// Start automatic compaction timer
insightsCompactionService.startCompactionTimer();
});
afterAll(async () => {
// Stop compaction timer and wait for any in-flight run to finish before the
// sibling afterAll terminates the DB connection.
await insightsCompactionService.stopCompactionTimer();
});
beforeEach(async () => {
@ -174,20 +164,25 @@ describe('Insights vs Workflow Statistics Integration', () => {
}
/**
* Helper to wait for insights to be compacted.
* Helper to compact insights and wait for the result.
*
* Polls until the compacted success count reaches the expected total. Waiting on the
* terminal count (rather than "some compacted data exists and raw is drained") avoids a
* race where events still buffered in the collection service haven't been flushed to
* InsightsRaw yet, so compaction runs on a partial set and the count comes up short.
* Runs a compaction pass per poll (the periodic cadence lives on the `insights-compaction`
* system task, which is not running here) and polls until the compacted success count
* reaches the expected total. Waiting on the terminal count (rather than "some compacted
* data exists and raw is drained") avoids a race where events still buffered in the
* collection service haven't been flushed to InsightsRaw yet, so compaction runs on a
* partial set and the count comes up short.
*/
async function waitForCompaction(
workflowId: string,
expectedSuccessCount: number,
timeout = 20000,
): Promise<void> {
const controller = new AbortController();
const start = Date.now();
while (Date.now() - start < timeout) {
await insightsCompactionService.compactInsights(controller.signal);
const compactedInsights = await insightsByPeriodRepository.find({
where: {
metadata: { workflowId },
@ -246,7 +241,7 @@ describe('Insights vs Workflow Statistics Integration', () => {
// Wait for workflow statistics to be recorded
await waitForStatistics(workflow.id, 10);
// Wait for automatic compaction to complete
// Compact and wait for the result
await waitForCompaction(workflow.id, 10);
// ============================================================