From 72960562ce7d2636480cac3fdb4a08cff3d5d5c6 Mon Sep 17 00:00:00 2001 From: Lorent Lempereur Date: Tue, 6 Oct 2026 11:10:53 +0000 Subject: [PATCH] refactor(core): Run the workflow statistics rollup as a durable system task (#40083) --- .../src/repositories/settings.repository.ts | 17 +++ .../@n8n/decorators/src/system-task/index.ts | 1 + .../decorators/src/system-task/system-task.ts | 6 +- .../mcp-registry-refresh.task.test.ts | 6 +- .../system-task-run-observer.test.ts | 19 +++ .../system-tasks/system-task-run-observer.ts | 10 +- ...workflow-statistics-rollup.service.test.ts | 65 +++++++- .../workflow-statistics-rollup.task.test.ts | 40 ++++- ...low-statistics.service.integration.test.ts | 76 ++++++++++ .../workflow-statistics-rollup.service.ts | 25 ++- .../workflow-statistics-rollup.task.ts | 40 +++-- .../services/workflow-statistics.service.ts | 23 ++- .../workflow-statistics-rollup.task.test.ts | 142 ++++++++++++++++++ 13 files changed, 414 insertions(+), 56 deletions(-) create mode 100644 packages/cli/test/integration/workflow-statistics-rollup.task.test.ts diff --git a/packages/@n8n/db/src/repositories/settings.repository.ts b/packages/@n8n/db/src/repositories/settings.repository.ts index 66532d201cf..ad062bf44a2 100644 --- a/packages/@n8n/db/src/repositories/settings.repository.ts +++ b/packages/@n8n/db/src/repositories/settings.repository.ts @@ -70,6 +70,23 @@ export class SettingsRepository extends BaseRepository { return (result.affected ?? 0) > 0; } + /** + * Stores the value, not loaded on startup, only when the key does not exist yet. + * Returns whether the stored value is this value, so exactly one of several + * concurrent callers with different values wins. + */ + async insertIfAbsent(key: string, value: string): Promise { + await this.manager + .createQueryBuilder() + .insert() + .into(Settings) + .values({ key, value, loadOnStartup: false }) + .orIgnore() + .execute(); + const stored = await this.findByKey(key); + return stored?.value === value; + } + async findByKeys(keys: string[]): Promise { return await this.findBy({ key: In(keys) }); } diff --git a/packages/@n8n/decorators/src/system-task/index.ts b/packages/@n8n/decorators/src/system-task/index.ts index cd7659f231e..ff0c42795ed 100644 --- a/packages/@n8n/decorators/src/system-task/index.ts +++ b/packages/@n8n/decorators/src/system-task/index.ts @@ -10,6 +10,7 @@ export { export type { SystemTaskClass, SystemTaskEffects, + SystemTaskRunContext, SystemTaskRunOptions, SystemTaskSchedule, } from './system-task'; diff --git a/packages/@n8n/decorators/src/system-task/system-task.ts b/packages/@n8n/decorators/src/system-task/system-task.ts index 320137b63ae..e07a0119414 100644 --- a/packages/@n8n/decorators/src/system-task/system-task.ts +++ b/packages/@n8n/decorators/src/system-task/system-task.ts @@ -81,7 +81,11 @@ export interface SystemTask { * An idempotent durable run that settles after a lease abort counts as a failed * attempt while its claim still matches, and retries only while attempts remain. */ - run(signal: AbortSignal): Promise; + run(signal: AbortSignal, context: SystemTaskRunContext): Promise; +} + +export interface SystemTaskRunContext { + readonly durable: boolean; } /** How a task's occurrences are retried and how late they may still run. */ diff --git a/packages/cli/src/modules/mcp-registry/__tests__/mcp-registry-refresh.task.test.ts b/packages/cli/src/modules/mcp-registry/__tests__/mcp-registry-refresh.task.test.ts index 0120827c477..b039500e470 100644 --- a/packages/cli/src/modules/mcp-registry/__tests__/mcp-registry-refresh.task.test.ts +++ b/packages/cli/src/modules/mcp-registry/__tests__/mcp-registry-refresh.task.test.ts @@ -23,7 +23,7 @@ describe('McpRegistryRefreshTask', () => { it('should refresh the registry on run and hand it the abort signal', async () => { const { signal } = new AbortController(); - await task.run(signal); + await task.run(signal, { durable: true }); expect(mcpRegistryService.refreshFromApi).toHaveBeenCalledExactlyOnceWith(signal); }); @@ -31,6 +31,8 @@ describe('McpRegistryRefreshTask', () => { it('should let a failed refresh propagate to the runner', async () => { mcpRegistryService.refreshFromApi.mockRejectedValue(new Error('api down')); - await expect(task.run(new AbortController().signal)).rejects.toThrow('api down'); + await expect(task.run(new AbortController().signal, { durable: true })).rejects.toThrow( + 'api down', + ); }); }); diff --git a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-run-observer.test.ts b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-run-observer.test.ts index fc4fc70deb8..8ff2890e114 100644 --- a/packages/cli/src/scheduling/system-tasks/__tests__/system-task-run-observer.test.ts +++ b/packages/cli/src/scheduling/system-tasks/__tests__/system-task-run-observer.test.ts @@ -36,6 +36,25 @@ describe('observeSystemTaskRun', () => { throw error; }; + it.each([ + { mode: 'durable' as const, durable: true }, + { mode: 'leader_timer' as const, durable: false }, + { mode: 'instance_timer' as const, durable: false }, + ])('tells a $mode run whether it is durable', async ({ mode, durable }) => { + const run = vi.fn(resolves); + const signal = new AbortController().signal; + + await observeSystemTaskRun( + mock(), + setupTracing().tracing, + taskThat(run), + mode, + signal, + ); + + expect(run).toHaveBeenCalledExactlyOnceWith(signal, { durable }); + }); + it('settles a run that throws synchronously as a failure, rather than rejecting', async () => { const eventService = mock(); const error = new Error('failed'); diff --git a/packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts b/packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts index aca7bdcc23c..7ccbedfc1dd 100644 --- a/packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts +++ b/packages/cli/src/scheduling/system-tasks/system-task-run-observer.ts @@ -19,8 +19,12 @@ const SYSTEM_TASK_ATTRIBUTES = { * tick, so an abort that follows the call cannot preempt the run, and rejects * rather than throws if `run` throws synchronously. */ -async function startRun(task: Pick, signal: AbortSignal): Promise { - await task.run(signal); +async function startRun( + task: Pick, + signal: AbortSignal, + mode: SystemTaskMode, +): Promise { + await task.run(signal, { durable: mode === 'durable' }); } /** How one observed run ended. A rejected run carries its error. */ @@ -68,7 +72,7 @@ export async function observeSystemTaskRun( return await startSpan(spanOptions, async (span: Span) => { const startedAt = performance.now(); emitSystemTaskMetric(eventService, 'system-task-run-started', { name, mode }); - const outcome = await startRun(task, runSignal).then( + const outcome = await startRun(task, runSignal, mode).then( (): SystemTaskRunOutcome => ({ result: abortResult() ?? 'success', rejected: false, diff --git a/packages/cli/src/services/__tests__/workflow-statistics-rollup.service.test.ts b/packages/cli/src/services/__tests__/workflow-statistics-rollup.service.test.ts index 2ba5c77a588..ee736c61389 100644 --- a/packages/cli/src/services/__tests__/workflow-statistics-rollup.service.test.ts +++ b/packages/cli/src/services/__tests__/workflow-statistics-rollup.service.test.ts @@ -30,7 +30,8 @@ describe('WorkflowStatisticsRollupService', () => { const rollup = async ( service: WorkflowStatisticsRollupService, signal = new AbortController().signal, - ) => await service.rollup(signal); + runBudgetMs?: number, + ) => await service.rollup(signal, runBudgetMs); const batchOf = (increments: number): RollupResult => ({ increments, firstOccurrences: [] }); @@ -107,6 +108,20 @@ describe('WorkflowStatisticsRollupService', () => { expect(dbLockService.tryWithLock).toHaveBeenCalledTimes(2); }); + it('drains beyond the task interval when no run budget is supplied', async () => { + const { service, dbLockService } = makeService(); + for (let i = 0; i < 24; i++) { + dbLockService.tryWithLock.mockResolvedValueOnce(batchOf(5000)); + } + dbLockService.tryWithLock.mockResolvedValueOnce(batchOf(12)); + + const run = rollup(service); + await vi.runAllTimersAsync(); + await run; + + expect(dbLockService.tryWithLock).toHaveBeenCalledTimes(25); + }); + it('stops at once when aborted during the pause between batches', async () => { const { service, dbLockService } = makeService(); dbLockService.tryWithLock.mockResolvedValue(batchOf(5000)); @@ -120,12 +135,32 @@ describe('WorkflowStatisticsRollupService', () => { expect(dbLockService.tryWithLock).toHaveBeenCalledTimes(1); }); + it('stops an unbudgeted drain when aborted past the task interval', async () => { + const { service, dbLockService } = makeService(); + dbLockService.tryWithLock.mockResolvedValue(batchOf(5000)); + const controller = new AbortController(); + + let settled = false; + void rollup(service, controller.signal).then(() => (settled = true)); + await vi.advanceTimersByTimeAsync(10_000); + expect(settled).toBe(false); + const batchesBeforeAbort = dbLockService.tryWithLock.mock.calls.length; + + controller.abort(); + await vi.advanceTimersByTimeAsync(0); + + expect(settled).toBe(true); + expect(batchesBeforeAbort).toBeGreaterThan(20); + expect(dbLockService.tryWithLock).toHaveBeenCalledTimes(batchesBeforeAbort); + expect(vi.getTimerCount()).toBe(0); + }); + it('stops once the run budget is spent, leaving the backlog to the next run', async () => { const { service, dbLockService } = makeService(); dbLockService.tryWithLock.mockResolvedValue(batchOf(5000)); let settled = false; - const run = rollup(service).then(() => (settled = true)); + const run = rollup(service, undefined, 4000).then(() => (settled = true)); await vi.advanceTimersByTimeAsync(4000); expect(settled).toBe(true); @@ -138,7 +173,7 @@ describe('WorkflowStatisticsRollupService', () => { dbLockService.tryWithLock.mockResolvedValue(batchOf(5000)); let settled = false; - void rollup(service).then(() => (settled = true)); + void rollup(service, undefined, 4000).then(() => (settled = true)); await vi.advanceTimersByTimeAsync(3750); expect(settled).toBe(true); @@ -152,13 +187,35 @@ describe('WorkflowStatisticsRollupService', () => { }); let settled = false; - void rollup(service).then(() => (settled = true)); + void rollup(service, undefined, 4000).then(() => (settled = true)); await vi.advanceTimersByTimeAsync(3250); expect(settled).toBe(true); expect(dbLockService.tryWithLock).toHaveBeenCalledTimes(2); }); + it('uses the run budget supplied by the caller', async () => { + const { service, dbLockService } = makeService(); + dbLockService.tryWithLock.mockResolvedValue(batchOf(5000)); + + let settled = false; + void rollup(service, undefined, 1000).then(() => (settled = true)); + await vi.advanceTimersByTimeAsync(750); + + expect(settled).toBe(true); + expect(dbLockService.tryWithLock).toHaveBeenCalledTimes(4); + }); + + it('folds one batch when the run budget is zero', async () => { + const { service, dbLockService } = makeService(); + dbLockService.tryWithLock.mockResolvedValue(batchOf(5000)); + + await rollup(service, undefined, 0); + + expect(dbLockService.tryWithLock).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + it('does not fold when the signal is already aborted', async () => { const { service, dbLockService } = makeService(); diff --git a/packages/cli/src/services/__tests__/workflow-statistics-rollup.task.test.ts b/packages/cli/src/services/__tests__/workflow-statistics-rollup.task.test.ts index ed47c281cdd..2fe8ec64d03 100644 --- a/packages/cli/src/services/__tests__/workflow-statistics-rollup.task.test.ts +++ b/packages/cli/src/services/__tests__/workflow-statistics-rollup.task.test.ts @@ -1,29 +1,53 @@ +import type { SchedulerConfig } from '@n8n/config'; import { mock } from 'vitest-mock-extended'; import type { WorkflowStatisticsRollupService } from '../workflow-statistics-rollup.service'; import { WorkflowStatisticsRollupTask } from '../workflow-statistics-rollup.task'; describe('WorkflowStatisticsRollupTask', () => { - let rollupService = mock(); - let task = new WorkflowStatisticsRollupTask(rollupService); + const rollupService = mock(); + const makeTask = (config: Partial = {}) => + new WorkflowStatisticsRollupTask( + rollupService, + mock({ leaseDurationSeconds: 60, ...config }), + ); beforeEach(() => { - rollupService = mock(); - task = new WorkflowStatisticsRollupTask(rollupService); + vi.clearAllMocks(); }); it('should declare a fixed 5 second cadence on the leader', () => { + const task = makeTask(); + expect(task.name).toBe('workflow-statistics-rollup'); expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 5 }); expect(task.effects).toBe('idempotent'); - expect(task.placement).toEqual({ scope: 'cluster', durable: false, runOnTakeover: true }); + expect(task.placement).toEqual({ scope: 'cluster', durable: true, runOnTakeover: true }); }); - it('should pass the run signal to the rollup', async () => { + it.each([ + { leaseDurationSeconds: 60, runBudgetMs: 4000 }, + { leaseDurationSeconds: 5, runBudgetMs: 4000 }, + { leaseDurationSeconds: 2, runBudgetMs: 1000 }, + { leaseDurationSeconds: 1, runBudgetMs: 0 }, + ])( + 'should pass the signal and a $runBudgetMs ms budget to a durable run with a $leaseDurationSeconds second lease', + async ({ leaseDurationSeconds, runBudgetMs }) => { + const task = makeTask({ leaseDurationSeconds }); + const signal = new AbortController().signal; + + await task.run(signal, { durable: true }); + + expect(rollupService.rollup).toHaveBeenCalledExactlyOnceWith(signal, runBudgetMs); + }, + ); + + it('should pass a budget below the interval to an in-memory run, whatever the lease', async () => { + const task = makeTask({ leaseDurationSeconds: 1 }); const signal = new AbortController().signal; - await task.run(signal); + await task.run(signal, { durable: false }); - expect(rollupService.rollup).toHaveBeenCalledWith(signal); + expect(rollupService.rollup).toHaveBeenCalledExactlyOnceWith(signal, 4000); }); }); diff --git a/packages/cli/src/services/__tests__/workflow-statistics.service.integration.test.ts b/packages/cli/src/services/__tests__/workflow-statistics.service.integration.test.ts index 21dd6cc6a1a..a3d8c78bc84 100644 --- a/packages/cli/src/services/__tests__/workflow-statistics.service.integration.test.ts +++ b/packages/cli/src/services/__tests__/workflow-statistics.service.integration.test.ts @@ -17,6 +17,7 @@ import { type EntityManager, type EntityMetadata, } from '@n8n/typeorm'; +import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import { mock } from 'vitest-mock-extended'; import { @@ -722,6 +723,81 @@ describe('WorkflowStatisticsService', () => { }); }); + describe('concurrent first occurrences', () => { + const makeService = () => { + const workflowRepositoryNoErrorWorkflows = mock(); + ( + workflowRepositoryNoErrorWorkflows as unknown as { + hasAnyWorkflowsWithErrorWorkflow: Mock; + } + ).hasAnyWorkflowsWithErrorWorkflow.mockResolvedValue(false); + return new WorkflowStatisticsService( + mock(), + workflowStatisticsRepository, + Container.get(OwnershipService), + userService, + Container.get(EventService), + Container.get(SettingsRepository), + workflowRepositoryNoErrorWorkflows, + Container.get(DatabaseConfig), + ); + }; + + test('emits instance-first-production-workflow-failed once for concurrent first failures', async () => { + const service = makeService(); + const otherWorkflow = await createWorkflow({}, user); + const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); + const settingsRepository = Container.get(SettingsRepository); + const findByKey = settingsRepository.findByKey.bind(settingsRepository); + const bothReadsCompleted = createDeferredPromise(); + let completedReads = 0; + + // Both calls must read the missing setting before either call can write it. + const readBeforeWrite: SettingsRepository['findByKey'] = async (...args) => { + const setting = await findByKey(...args); + if (++completedReads === 2) { + bothReadsCompleted.resolve(); + } + await bothReadsCompleted.promise; + return setting; + }; + vi.spyOn(settingsRepository, 'findByKey') + .mockImplementationOnce(readBeforeWrite) + .mockImplementationOnce(readBeforeWrite); + + await Promise.all( + [workflow, otherWorkflow].map( + async (wf) => + await service.emitFirstOccurrenceEvent( + StatisticsNames.productionError, + wf.id, + wf.name, + Date.now(), + ), + ), + ); + + const failureEvents = emitSpy.mock.calls.filter( + ([eventName]) => eventName === 'instance-first-production-workflow-failed', + ); + expect(failureEvents).toHaveLength(1); + }); + + test('keeps the first value when the instance activation key is inserted twice', async () => { + const settingsRepository = Container.get(SettingsRepository); + + await expect( + settingsRepository.insertIfAbsent(INSTANCE_ACTIVATED_SETTINGS_KEY, 'a'), + ).resolves.toBe(true); + await expect( + settingsRepository.insertIfAbsent(INSTANCE_ACTIVATED_SETTINGS_KEY, 'b'), + ).resolves.toBe(false); + + const row = await settingsRepository.findByKey(INSTANCE_ACTIVATED_SETTINGS_KEY); + expect(row?.value).toBe('a'); + }); + }); + // The fold is Postgres-only (raw CTE). These exercise its mechanics directly via the repository. describe('rollupIncrements (Postgres append path)', () => { // The delta table is cleared by the outer beforeEach via testDb.truncate. diff --git a/packages/cli/src/services/workflow-statistics-rollup.service.ts b/packages/cli/src/services/workflow-statistics-rollup.service.ts index 680ed4ba176..5e8749eb018 100644 --- a/packages/cli/src/services/workflow-statistics-rollup.service.ts +++ b/packages/cli/src/services/workflow-statistics-rollup.service.ts @@ -1,5 +1,4 @@ import { Logger } from '@n8n/backend-common'; -import { Time } from '@n8n/constants'; import { DbLock, DbLockService, WorkflowStatisticsRepository } from '@n8n/db'; import { Service } from '@n8n/di'; import { sleep } from '@n8n/utils/sleep'; @@ -15,11 +14,6 @@ const BATCH_SIZE = 5000; /** Pause between full batches, i.e. while backlog remains. */ const BATCH_DELAY_MS = 250; -export const ROLLUP_INTERVAL_SECONDS = 5; - -/** Leaves a margin below the task interval for a batch slower than the one before. */ -const RUN_BUDGET_MS = (ROLLUP_INTERVAL_SECONDS - 1) * Time.seconds.toMilliseconds; - /** Consecutive lock skips after which to warn that the lock is persistently held elsewhere. */ const SKIP_WARN_THRESHOLD = 5; @@ -45,10 +39,11 @@ export class WorkflowStatisticsRollupService { /** * Fold batches until one comes back partial, the signal aborts, or the run - * budget has no room left for another batch. + * budget has no room left for another batch. The optional budget is in + * milliseconds. It limits additional batches, but does not interrupt a batch. */ - async rollup(signal: AbortSignal): Promise { - const deadline = Date.now() + RUN_BUDGET_MS; + async rollup(signal: AbortSignal, runBudgetMs = Number.POSITIVE_INFINITY): Promise { + const deadline = Date.now() + runBudgetMs; let batchStartedAt = Date.now(); while ( !signal.aborted && @@ -78,7 +73,7 @@ export class WorkflowStatisticsRollupService { } } - /** Fold a batch under an advisory lock. Returns null if another instance holds the lock. */ + /** Fold a batch under an advisory lock. Returns null if another run or process holds the lock. */ private async foldBatch(): Promise { try { const result = await this.dbLockService.tryWithLock( @@ -89,7 +84,7 @@ export class WorkflowStatisticsRollupService { return result; } catch (error) { if (error instanceof OperationalError) { - this.registerLockSkip(); // another instance holds the lock + this.registerLockSkip(); // another run or process holds the lock return null; } throw error; @@ -97,9 +92,9 @@ export class WorkflowStatisticsRollupService { } /** - * Occasional skips are expected around leader transitions; persistent skips suggest a process - * outside this deployment holds the lock, e.g. a second n8n instance sharing this database - * (advisory locks are not schema- or table-prefix-scoped). + * Occasional skips are expected around leader transitions and when a slow run overlaps the next + * one; persistent skips suggest a process outside this deployment holds the lock, e.g. a second + * n8n instance sharing this database (advisory locks are not schema- or table-prefix-scoped). */ private registerLockSkip() { this.consecutiveLockSkips++; @@ -108,7 +103,7 @@ export class WorkflowStatisticsRollupService { if (this.consecutiveLockSkips % SKIP_WARN_THRESHOLD !== 0) return; this.logger.warn( - 'Workflow statistics rollup repeatedly skipped: lock held by another process', + 'Workflow statistics rollup repeatedly skipped: lock held by another run or process', { consecutiveLockSkips: this.consecutiveLockSkips, totalLockSkips: this.totalLockSkips, diff --git a/packages/cli/src/services/workflow-statistics-rollup.task.ts b/packages/cli/src/services/workflow-statistics-rollup.task.ts index 95bf782be9e..085d74910aa 100644 --- a/packages/cli/src/services/workflow-statistics-rollup.task.ts +++ b/packages/cli/src/services/workflow-statistics-rollup.task.ts @@ -1,10 +1,19 @@ +import { SchedulerConfig } from '@n8n/config'; +import { Time } from '@n8n/constants'; import { intervalFromSeconds, SystemTask } from '@n8n/decorators'; -import type { SystemTaskEffects, SystemTaskPlacement, SystemTaskSchedule } from '@n8n/decorators'; +import type { + SystemTaskEffects, + SystemTaskPlacement, + SystemTaskRunContext, + SystemTaskSchedule, +} from '@n8n/decorators'; -import { - ROLLUP_INTERVAL_SECONDS, - WorkflowStatisticsRollupService, -} from './workflow-statistics-rollup.service'; +import { WorkflowStatisticsRollupService } from './workflow-statistics-rollup.service'; + +const ROLLUP_INTERVAL_SECONDS = 5; + +/** Ends a run before the next occurrence, so a backlog does not cause overlap skips. */ +const IN_MEMORY_RUN_BUDGET_MS = (ROLLUP_INTERVAL_SECONDS - 1) * Time.seconds.toMilliseconds; /** * Folds the pending workflow statistics increments into the counters, and @@ -20,14 +29,27 @@ export class WorkflowStatisticsRollupTask implements SystemTask { readonly placement: SystemTaskPlacement = { scope: 'cluster', - durable: false, + durable: true, /** Increments pile up while no instance is the leader, so a backlog is waiting. */ runOnTakeover: true, }; - constructor(private readonly rollupService: WorkflowStatisticsRollupService) {} + private readonly durableRunBudgetMs: number; - async run(signal: AbortSignal): Promise { - await this.rollupService.rollup(signal); + constructor( + private readonly rollupService: WorkflowStatisticsRollupService, + schedulerConfig: SchedulerConfig, + ) { + // Leave one second for a slower batch. The durable lease is not renewed during a run. + this.durableRunBudgetMs = + (Math.min(ROLLUP_INTERVAL_SECONDS, schedulerConfig.leaseDurationSeconds) - 1) * + Time.seconds.toMilliseconds; + } + + async run(signal: AbortSignal, { durable }: SystemTaskRunContext): Promise { + await this.rollupService.rollup( + signal, + durable ? this.durableRunBudgetMs : IN_MEMORY_RUN_BUDGET_MS, + ); } } diff --git a/packages/cli/src/services/workflow-statistics.service.ts b/packages/cli/src/services/workflow-statistics.service.ts index 9665e0581bf..a54ce777972 100644 --- a/packages/cli/src/services/workflow-statistics.service.ts +++ b/packages/cli/src/services/workflow-statistics.service.ts @@ -264,17 +264,10 @@ export class WorkflowStatisticsService extends TypedEmitter { - const alreadyActivated = await this.settingsRepository.findByKey( + await this.settingsRepository.insertIfAbsent( INSTANCE_ACTIVATED_SETTINGS_KEY, + JSON.stringify({ workflowId, projectId, userId, timestamp: activatedAt }), ); - - if (alreadyActivated) return; - - await this.settingsRepository.save({ - key: INSTANCE_ACTIVATED_SETTINGS_KEY, - value: JSON.stringify({ workflowId, projectId, userId, timestamp: activatedAt }), - loadOnStartup: false, - }); } private async emitInstanceFirstProductionWorkflowFailed( @@ -302,16 +295,18 @@ export class WorkflowStatisticsService extends TypedEmitter { + const signal = new AbortController().signal; + let repository: WorkflowStatisticsRepository; + let statisticsService: ReturnType>; + let task: WorkflowStatisticsRollupTask; + let deltaTable: string; + + beforeAll(async () => { + // One connection per run, so a run can meet the lock that another run holds. + Container.get(GlobalConfig).database.postgresdb.poolSize = OVERLAPPING_RUNS; + await testDb.init(); + repository = Container.get(WorkflowStatisticsRepository); + deltaTable = `${Container.get(GlobalConfig).database.tablePrefix}workflow_statistics_delta`; + }); + + beforeEach(() => { + statisticsService = mock(); + task = new WorkflowStatisticsRollupTask( + new WorkflowStatisticsRollupService( + mockLogger(), + mock(), + Container.get(DbLockService), + repository, + statisticsService, + ), + mock({ leaseDurationSeconds: 60 }), + ); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + await testDb.truncate(['WorkflowStatistics', 'WorkflowStatisticsDelta']); + }); + + afterAll(async () => await testDb.terminate()); + + async function seedIncrements(): Promise { + for (const workflowId of WORKFLOW_IDS) { + await repository.query( + `INSERT INTO ${deltaTable} ("workflowId", "name", "rootCountDelta", "workflowName") + SELECT $1, $2, 1, 'wf' FROM generate_series(1, $3)`, + [workflowId, StatisticsNames.productionSuccess, ROWS_PER_WORKFLOW], + ); + } + } + + async function pendingIncrements(): Promise { + const [{ count }]: Array<{ count: number }> = await repository.query( + `SELECT COUNT(*)::int AS count FROM ${deltaTable}`, + ); + return count; + } + + /** Holds the first fold inside the lock until another run is refused the lock, or a timeout passes. */ + function holdFirstFoldUntilLockSkip(): { lockSkips: () => number } { + const dbLockService = Container.get(DbLockService); + const tryWithLock = dbLockService.tryWithLock.bind(dbLockService); + const rollupIncrements = repository.rollupIncrements.bind(repository); + let lockSkips = 0; + let onLockSkip = () => {}; + const lockSkipped = new Promise((resolve) => { + onLockSkip = resolve; + }); + + vi.spyOn(dbLockService, 'tryWithLock').mockImplementation(async (lockId, fn, options) => { + try { + return await tryWithLock(lockId, fn, options); + } catch (error) { + if (error instanceof OperationalError) { + lockSkips++; + onLockSkip(); + } + throw error; + } + }); + vi.spyOn(repository, 'rollupIncrements').mockImplementationOnce(async (tx, batchSize) => { + await Promise.race([lockSkipped, sleep(LOCK_SKIP_TIMEOUT_MS)]); + return await rollupIncrements(tx, batchSize); + }); + + return { lockSkips: () => lockSkips }; + } + + it('should fold each increment once and fire each milestone once when runs overlap', async () => { + await seedIncrements(); + const { lockSkips } = holdFirstFoldUntilLockSkip(); + + await Promise.all( + Array.from( + { length: OVERLAPPING_RUNS }, + async () => await task.run(signal, { durable: true }), + ), + ); + expect(lockSkips()).toBeGreaterThan(0); + // A run that lost the lock stops early, so finish any backlog it left. + while ((await pendingIncrements()) > 0) { + await task.run(signal, { durable: true }); + } + + const counters = await repository.findBy({ name: StatisticsNames.productionSuccess }); + expect( + counters.map(({ workflowId, count, rootCount }) => ({ workflowId, count, rootCount })), + ).toEqual( + expect.arrayContaining( + WORKFLOW_IDS.map((workflowId) => ({ + workflowId, + count: ROWS_PER_WORKFLOW, + rootCount: ROWS_PER_WORKFLOW, + })), + ), + ); + expect(counters).toHaveLength(WORKFLOW_IDS.length); + + const milestoneWorkflowIds = statisticsService.emitFirstOccurrenceEvent.mock.calls.map( + ([, workflowId]) => workflowId, + ); + expect(milestoneWorkflowIds.sort()).toEqual([...WORKFLOW_IDS].sort()); + }); +});