refactor(core): Run the workflow statistics rollup as a durable system task (#40083)

packages/@n8n/db/src/repositories/settings.repository.ts

@@ -70,6 +70,23 @@ export class SettingsRepository extends BaseRepository<Settings> {
 		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<boolean> {
+		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<Settings[]> {
 		return await this.findBy({ key: In(keys) });
 	}

packages/@n8n/decorators/src/system-task/index.ts

@@ -10,6 +10,7 @@ export {
 export type {
 	SystemTaskClass,
 	SystemTaskEffects,
+	SystemTaskRunContext,
 	SystemTaskRunOptions,
 	SystemTaskSchedule,
 } from './system-task';

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<void>;
+	run(signal: AbortSignal, context: SystemTaskRunContext): Promise<void>;
+}
+
+export interface SystemTaskRunContext {
+	readonly durable: boolean;
 }
 
 /** How a task's occurrences are retried and how late they may still run. */

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

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<EventService>(),
+			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<EventService>();
 		const error = new Error('failed');

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<SystemTask, 'run'>, signal: AbortSignal): Promise<void> {
-	await task.run(signal);
+async function startRun(
+	task: Pick<SystemTask, 'run'>,
+	signal: AbortSignal,
+	mode: SystemTaskMode,
+): Promise<void> {
+	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,

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();
 

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<WorkflowStatisticsRollupService>();
-	let task = new WorkflowStatisticsRollupTask(rollupService);
+	const rollupService = mock<WorkflowStatisticsRollupService>();
+	const makeTask = (config: Partial<SchedulerConfig> = {}) =>
+		new WorkflowStatisticsRollupTask(
+			rollupService,
+			mock<SchedulerConfig>({ leaseDurationSeconds: 60, ...config }),
+		);
 
 	beforeEach(() => {
-		rollupService = mock<WorkflowStatisticsRollupService>();
-		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);
 	});
 });

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<WorkflowRepository>();
+				(
+					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.

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<void> {
-		const deadline = Date.now() + RUN_BUDGET_MS;
+	async rollup(signal: AbortSignal, runBudgetMs = Number.POSITIVE_INFINITY): Promise<void> {
+		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<RollupResult | null> {
 		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,

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<void> {
-		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<void> {
+		await this.rollupService.rollup(
+			signal,
+			durable ? this.durableRunBudgetMs : IN_MEMORY_RUN_BUDGET_MS,
+		);
 	}
 }

packages/cli/src/services/workflow-statistics.service.ts

@@ -264,17 +264,10 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
 		userId: string | null,
 		activatedAt: number,
 	): Promise<void> {
-		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<WorkflowStatisticsEv
 
 		owner ??= await this.ownershipService.getInstanceOwner();
 
-		await this.settingsRepository.save({
-			key: 'instance.firstProductionFailure',
-			value: JSON.stringify({
+		const isFirstFailure = await this.settingsRepository.insertIfAbsent(
+			'instance.firstProductionFailure',
+			JSON.stringify({
 				workflowId,
 				projectId: project.id,
 				userId: owner.id,
 				timestamp: timestampMs,
 			}),
-			loadOnStartup: false,
-		});
+		);
+		if (!isFirstFailure) {
+			return;
+		}
 
 		this.eventService.emit('instance-first-production-workflow-failed', {
 			projectId: project.id,

packages/cli/test/integration/workflow-statistics-rollup.task.test.ts

@@ -0,0 +1,142 @@
+import { mockLogger, testDb } from '@n8n/backend-test-utils';
+import { GlobalConfig } from '@n8n/config';
+import type { SchedulerConfig } from '@n8n/config';
+import { DbLockService, StatisticsNames, WorkflowStatisticsRepository } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { sleep } from '@n8n/utils/sleep';
+import { mock } from 'vitest-mock-extended';
+import type { ErrorReporter } from 'n8n-core';
+import { OperationalError } from 'n8n-workflow';
+
+import type { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
+import { WorkflowStatisticsRollupService } from '@/services/workflow-statistics-rollup.service';
+import { WorkflowStatisticsRollupTask } from '@/services/workflow-statistics-rollup.task';
+
+const isPostgres = process.env.DB_TYPE === 'postgresdb';
+
+const OVERLAPPING_RUNS = 4;
+const WORKFLOW_IDS = ['wf-rollup-1', 'wf-rollup-2', 'wf-rollup-3'];
+/** More than one 5000-row batch in total, so the runs contend for several folds. */
+const ROWS_PER_WORKFLOW = 2_000;
+const LOCK_SKIP_TIMEOUT_MS = 5_000;
+
+/** The delta table exists on Postgres only. */
+describe.skipIf(!isPostgres)('WorkflowStatisticsRollupTask', () => {
+	const signal = new AbortController().signal;
+	let repository: WorkflowStatisticsRepository;
+	let statisticsService: ReturnType<typeof mock<WorkflowStatisticsService>>;
+	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<WorkflowStatisticsService>();
+		task = new WorkflowStatisticsRollupTask(
+			new WorkflowStatisticsRollupService(
+				mockLogger(),
+				mock<ErrorReporter>(),
+				Container.get(DbLockService),
+				repository,
+				statisticsService,
+			),
+			mock<SchedulerConfig>({ leaseDurationSeconds: 60 }),
+		);
+	});
+
+	afterEach(async () => {
+		vi.restoreAllMocks();
+		await testDb.truncate(['WorkflowStatistics', 'WorkflowStatisticsDelta']);
+	});
+
+	afterAll(async () => await testDb.terminate());
+
+	async function seedIncrements(): Promise<void> {
+		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<number> {
+		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<void>((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());
+	});
+});