From 52befceb667896da34bbe5e338331bad7e415703 Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Tue, 8 Sep 2026 07:15:14 +0000 Subject: [PATCH] refactor(core): Run insights compaction and pruning as system tasks (no-changelog) (#37543) Co-authored-by: Veljko Terzic --- packages/cli/eslint.config.mjs | 1 - ...hts-compaction.service.integration.test.ts | 305 +++++++----------- .../insights-compaction.task.test.ts | 29 ++ .../insights-pruning.service.test.ts | 103 ++---- .../__tests__/insights-pruning.task.test.ts | 28 ++ .../__tests__/insights.config.test.ts | 42 +++ .../__tests__/insights.module.test.ts | 13 +- .../insights.service.integration.test.ts | 91 +----- .../__tests__/insights.service.test.ts | 8 - .../insights/__tests__/whole-seconds.test.ts | 12 + .../insights/insights-compaction.service.ts | 83 ++--- .../insights/insights-compaction.task.ts | 36 +++ .../insights/insights-pruning.service.ts | 58 +--- .../modules/insights/insights-pruning.task.ts | 38 +++ .../src/modules/insights/insights.config.ts | 8 +- .../src/modules/insights/insights.module.ts | 6 + .../src/modules/insights/insights.service.ts | 20 -- .../cli/src/modules/insights/whole-seconds.ts | 4 + ...vs-workflow-statistics.integration.test.ts | 29 +- 19 files changed, 415 insertions(+), 499 deletions(-) create mode 100644 packages/cli/src/modules/insights/__tests__/insights-compaction.task.test.ts create mode 100644 packages/cli/src/modules/insights/__tests__/insights-pruning.task.test.ts create mode 100644 packages/cli/src/modules/insights/__tests__/insights.config.test.ts create mode 100644 packages/cli/src/modules/insights/__tests__/whole-seconds.test.ts create mode 100644 packages/cli/src/modules/insights/insights-compaction.task.ts create mode 100644 packages/cli/src/modules/insights/insights-pruning.task.ts create mode 100644 packages/cli/src/modules/insights/whole-seconds.ts diff --git a/packages/cli/eslint.config.mjs b/packages/cli/eslint.config.mjs index 8e846e427cd..f3826c5b986 100644 --- a/packages/cli/eslint.config.mjs +++ b/packages/cli/eslint.config.mjs @@ -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', diff --git a/packages/cli/src/modules/insights/__tests__/insights-compaction.service.integration.test.ts b/packages/cli/src/modules/insights/__tests__/insights-compaction.service.integration.test.ts index 19803462def..fa266356113 100644 --- a/packages/cli/src/modules/insights/__tests__/insights-compaction.service.integration.test.ts +++ b/packages/cli/src/modules/insights/__tests__/insights-compaction.service.integration.test.ts @@ -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(), - mock(), - mock({ - 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({ scoped: vi.fn().mockReturnThis(), debug: loggerDebug }); - const insightsCompactionService = new InsightsCompactionService( - mock(), - mock(), - mock({ - compactionIntervalMinutes: 1, - compactionBatchSize: 2, - compactionBatchDelayMilliseconds: 0, - compactionMaxBatchesPerRun: 0, - compactionMaxRuntimeSeconds: 0, - }), - logger, - ); - - let resolveRawToHour!: (numberOfCompactedRawData: number) => void; - const firstRawToHourPromise = new Promise((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(), - mock(), - mock({ - 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({ scoped: vi.fn().mockReturnThis() }); - const insightsCompactionService = new InsightsCompactionService( - mock(), - mock(), - mock({ - compactionBatchSize: 2, - compactionBatchDelayMilliseconds: 0, - compactionMaxBatchesPerRun: 0, - compactionMaxRuntimeSeconds: 0, - }), - logger, - ); - - let resolveRawToHour!: (numberOfCompactedRawData: number) => void; - const firstRawToHourPromise = new Promise((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({ 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); diff --git a/packages/cli/src/modules/insights/__tests__/insights-compaction.task.test.ts b/packages/cli/src/modules/insights/__tests__/insights-compaction.task.test.ts new file mode 100644 index 00000000000..b45570e770e --- /dev/null +++ b/packages/cli/src/modules/insights/__tests__/insights-compaction.task.test.ts @@ -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(); + 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); + }); +}); diff --git a/packages/cli/src/modules/insights/__tests__/insights-pruning.service.test.ts b/packages/cli/src/modules/insights/__tests__/insights-pruning.service.test.ts index 1a21edea8b4..a1f4646c41e 100644 --- a/packages/cli/src/modules/insights/__tests__/insights-pruning.service.test.ts +++ b/packages/cli/src/modules/insights/__tests__/insights-pruning.service.test.ts @@ -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(); 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({ - 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({ - 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'); }); }); }); diff --git a/packages/cli/src/modules/insights/__tests__/insights-pruning.task.test.ts b/packages/cli/src/modules/insights/__tests__/insights-pruning.task.test.ts new file mode 100644 index 00000000000..2f7e6bd4c09 --- /dev/null +++ b/packages/cli/src/modules/insights/__tests__/insights-pruning.task.test.ts @@ -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(); + 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); + }); +}); diff --git a/packages/cli/src/modules/insights/__tests__/insights.config.test.ts b/packages/cli/src/modules/insights/__tests__/insights.config.test.ts new file mode 100644 index 00000000000..52b57f67ada --- /dev/null +++ b/packages/cli/src/modules/insights/__tests__/insights.config.test.ts @@ -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); + }, + ); +}); diff --git a/packages/cli/src/modules/insights/__tests__/insights.module.test.ts b/packages/cli/src/modules/insights/__tests__/insights.module.test.ts index f384a4537b4..62e63f2eaae 100644 --- a/packages/cli/src/modules/insights/__tests__/insights.module.test.ts +++ b/packages/cli/src/modules/insights/__tests__/insights.module.test.ts @@ -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 diff --git a/packages/cli/src/modules/insights/__tests__/insights.service.integration.test.ts b/packages/cli/src/modules/insights/__tests__/insights.service.integration.test.ts index 8bb6f03427a..a92e4628f34 100644 --- a/packages/cli/src/modules/insights/__tests__/insights.service.integration.test.ts +++ b/packages/cli/src/modules/insights/__tests__/insights.service.integration.test.ts @@ -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; let realCollectionService: InsightsCollectionService; let initSpy: MockInstance; let shutdownSpy: MockInstance; beforeEach(() => { - compactionService = mock(); - pruningService = mock(); instanceSettings = mock({ instanceType: 'main', }); insightsService = new InsightsService( mock(), - compactionService, - pruningService, mock(), 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(); insightsService = new InsightsService( mock(), - mock(), - mock(), licenseStateMock, mock(), mockLogger(), @@ -1538,19 +1509,9 @@ describe('InsightsService (Integration)', () => { describe('shutdown', () => { let insightsService: InsightsService; - const mockCompactionService = mock({ - stopCompactionTimer: vi.fn(), - }); - - const mockPruningService = mock({ - stopPruningTimer: vi.fn(), - }); - beforeAll(() => { insightsService = new InsightsService( mock(), - mockCompactionService, - mockPruningService, mock(), mock({ 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((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; }); }); }); diff --git a/packages/cli/src/modules/insights/__tests__/insights.service.test.ts b/packages/cli/src/modules/insights/__tests__/insights.service.test.ts index d8b9258b385..6ad1a186664 100644 --- a/packages/cli/src/modules/insights/__tests__/insights.service.test.ts +++ b/packages/cli/src/modules/insights/__tests__/insights.service.test.ts @@ -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; - let mockCompactionService: MockProxy; - let mockPruningService: MockProxy; let mockLicenseState: MockProxy; let mockInstanceSettings: MockProxy; let mockWorkflowSharingService: MockProxy; @@ -34,8 +30,6 @@ describe('InsightsService', () => { vi.clearAllMocks(); mockInsightsByPeriodRepository = mock(); - mockCompactionService = mock(); - mockPruningService = mock(); mockLicenseState = mock(); mockInstanceSettings = mock(); mockWorkflowSharingService = mock(); @@ -43,8 +37,6 @@ describe('InsightsService', () => { insightsService = new InsightsService( mockInsightsByPeriodRepository, - mockCompactionService, - mockPruningService, mockLicenseState, mockInstanceSettings, mockLogger(), diff --git a/packages/cli/src/modules/insights/__tests__/whole-seconds.test.ts b/packages/cli/src/modules/insights/__tests__/whole-seconds.test.ts new file mode 100644 index 00000000000..c075e18a9d5 --- /dev/null +++ b/packages/cli/src/modules/insights/__tests__/whole-seconds.test.ts @@ -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); + }); +}); diff --git a/packages/cli/src/modules/insights/insights-compaction.service.ts b/packages/cli/src/modules/insights/insights-compaction.service.ts index 5c19fff8898..50290085b15 100644 --- a/packages/cli/src/modules/insights/insights-compaction.service.ts +++ b/packages/cli/src/modules/insights/insights-compaction.service.ts @@ -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 | 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. + } } /** diff --git a/packages/cli/src/modules/insights/insights-compaction.task.ts b/packages/cli/src/modules/insights/insights-compaction.task.ts new file mode 100644 index 00000000000..35a0dd534ac --- /dev/null +++ b/packages/cli/src/modules/insights/insights-compaction.task.ts @@ -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 { + await this.compactionService.compactInsights(signal); + } +} diff --git a/packages/cli/src/modules/insights/insights-pruning.service.ts b/packages/cli/src/modules/insights/insights-pruning.service.ts index 84d1d9ceb2c..a2d01e5ffee 100644 --- a/packages/cli/src/modules/insights/insights-pruning.service.ts +++ b/packages/cli/src/modules/insights/insights-pruning.service.ts @@ -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 } : {}, + ); } } diff --git a/packages/cli/src/modules/insights/insights-pruning.task.ts b/packages/cli/src/modules/insights/insights-pruning.task.ts new file mode 100644 index 00000000000..854bccd277c --- /dev/null +++ b/packages/cli/src/modules/insights/insights-pruning.task.ts @@ -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 { + await this.pruningService.pruneInsights(); + } +} diff --git a/packages/cli/src/modules/insights/insights.config.ts b/packages/cli/src/modules/insights/insights.config.ts index 7fbb76fa17f..73d6ba3802d 100644 --- a/packages/cli/src/modules/insights/insights.config.ts +++ b/packages/cli/src/modules/insights/insights.config.ts @@ -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; /** diff --git a/packages/cli/src/modules/insights/insights.module.ts b/packages/cli/src/modules/insights/insights.module.ts index 6f5c2183f78..c511a693d58 100644 --- a/packages/cli/src/modules/insights/insights.module.ts +++ b/packages/cli/src/modules/insights/insights.module.ts @@ -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'); diff --git a/packages/cli/src/modules/insights/insights.service.ts b/packages/cli/src/modules/insights/insights.service.ts index f43a1910ab7..bbc3db71da0 100644 --- a/packages/cli/src/modules/insights/insights.service.ts +++ b/packages/cli/src/modules/insights/insights.service.ts @@ -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(); } /** diff --git a/packages/cli/src/modules/insights/whole-seconds.ts b/packages/cli/src/modules/insights/whole-seconds.ts new file mode 100644 index 00000000000..101f1179d10 --- /dev/null +++ b/packages/cli/src/modules/insights/whole-seconds.ts @@ -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)); +} diff --git a/packages/cli/test/integration/insights/insights-vs-workflow-statistics.integration.test.ts b/packages/cli/test/integration/insights/insights-vs-workflow-statistics.integration.test.ts index 2d8c1a52370..4169818750e 100644 --- a/packages/cli/test/integration/insights/insights-vs-workflow-statistics.integration.test.ts +++ b/packages/cli/test/integration/insights/insights-vs-workflow-statistics.integration.test.ts @@ -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 { + 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); // ============================================================