diff --git a/docs/generated/postgres-schema/README.md b/docs/generated/postgres-schema/README.md index 85b57a01a79..037a9ea6806 100644 --- a/docs/generated/postgres-schema/README.md +++ b/docs/generated/postgres-schema/README.md @@ -119,6 +119,7 @@ Auto-generated from the PostgreSQL migrations in @n8n/db. Do not edit by hand. | [public.workflow_publish_history](public.workflow_publish_history.md) | 6 | | BASE TABLE | | [public.workflow_published_version](public.workflow_published_version.md) | 4 | | BASE TABLE | | [public.workflow_statistics](public.workflow_statistics.md) | 7 | | BASE TABLE | +| [public.workflow_statistics_delta](public.workflow_statistics_delta.md) | 6 | | BASE TABLE | | [public.workflows_tags](public.workflows_tags.md) | 2 | | BASE TABLE | ## Stored procedures and functions @@ -1372,6 +1373,14 @@ erDiagram varchar_36_ workflowId varchar_128_ workflowName } +"public.workflow_statistics_delta" { + timestamp_3__with_time_zone createdAt + bigint id + varchar_128_ name + smallint rootCountDelta + varchar_36_ workflowId + varchar_128_ workflowName +} "public.workflows_tags" { varchar_36_ tagId FK varchar_36_ workflowId FK diff --git a/docs/generated/postgres-schema/public.workflow_statistics_delta.md b/docs/generated/postgres-schema/public.workflow_statistics_delta.md new file mode 100644 index 00000000000..39cfaef6e34 --- /dev/null +++ b/docs/generated/postgres-schema/public.workflow_statistics_delta.md @@ -0,0 +1,49 @@ +# public.workflow_statistics_delta + +## Columns + +| Name | Type | Default | Nullable | Children | Parents | Comment | +| ---- | ---- | ------- | -------- | -------- | ------- | ------- | +| createdAt | timestamp(3) with time zone | CURRENT_TIMESTAMP | false | | | | +| id | bigint | | false | | | | +| name | varchar(128) | | false | | | | +| rootCountDelta | smallint | | false | | | | +| workflowId | varchar(36) | | false | | | | +| workflowName | varchar(128) | | true | | | | + +## Constraints + +| Name | Type | Definition | +| ---- | ---- | ---------- | +| workflow_statistics_delta_createdAt_not_null | n | NOT NULL "createdAt" | +| workflow_statistics_delta_id_not_null | n | NOT NULL id | +| workflow_statistics_delta_name_not_null | n | NOT NULL name | +| workflow_statistics_delta_pkey | PRIMARY KEY | PRIMARY KEY (id) | +| workflow_statistics_delta_rootCountDelta_not_null | n | NOT NULL "rootCountDelta" | +| workflow_statistics_delta_workflowId_not_null | n | NOT NULL "workflowId" | + +## Indexes + +| Name | Definition | +| ---- | ---------- | +| workflow_statistics_delta_pkey | CREATE UNIQUE INDEX workflow_statistics_delta_pkey ON public.workflow_statistics_delta USING btree (id) | + +## Relations + +```mermaid +erDiagram + + +"public.workflow_statistics_delta" { + timestamp_3__with_time_zone createdAt + bigint id + varchar_128_ name + smallint rootCountDelta + varchar_36_ workflowId + varchar_128_ workflowName +} +``` + +--- + +> Generated by [tbls](https://github.com/k1LoW/tbls) diff --git a/packages/@n8n/backend-test-utils/src/test-db.ts b/packages/@n8n/backend-test-utils/src/test-db.ts index 6488d289ecc..0676deb6f8d 100644 --- a/packages/@n8n/backend-test-utils/src/test-db.ts +++ b/packages/@n8n/backend-test-utils/src/test-db.ts @@ -172,7 +172,8 @@ type EntityName = | 'DynamicCredentialUserEntry' | 'TokenExchangeJti' | 'TrustedKeySourceEntity' - | 'TrustedKeyEntity'; + | 'TrustedKeyEntity' + | 'WorkflowStatisticsDelta'; /** * Truncate specific DB tables in a test DB. @@ -204,6 +205,16 @@ export async function truncate(entities: EntityName[]) { } for (const name of entities) { + // `workflow_statistics_delta` is a raw-SQL, Postgres-only table with no TypeORM entity, so it + // can't go through the repository, so we clear it directly. + if (name === 'WorkflowStatisticsDelta') { + const { type, tablePrefix } = Container.get(GlobalConfig).database; + if (type === 'postgresdb') { + const table = connection.driver.escape(`${tablePrefix}workflow_statistics_delta`); + await connection.query(`DELETE FROM ${table}`); + } + continue; + } await connection.getRepository(name).delete({}); } } diff --git a/packages/@n8n/config/src/configs/logging.config.ts b/packages/@n8n/config/src/configs/logging.config.ts index 1a100bae7e7..0f7bf27221f 100644 --- a/packages/@n8n/config/src/configs/logging.config.ts +++ b/packages/@n8n/config/src/configs/logging.config.ts @@ -22,6 +22,7 @@ export const LOG_SCOPES = [ 'task-runner-js', 'task-runner-py', 'insights', + 'workflow-statistics', 'workflow-activation', 'ssh-client', 'data-table', diff --git a/packages/@n8n/db/src/migrations/postgresdb/1784000000043-CreateWorkflowStatisticsDeltaTable.ts b/packages/@n8n/db/src/migrations/postgresdb/1784000000043-CreateWorkflowStatisticsDeltaTable.ts new file mode 100644 index 00000000000..0b51c770aad --- /dev/null +++ b/packages/@n8n/db/src/migrations/postgresdb/1784000000043-CreateWorkflowStatisticsDeltaTable.ts @@ -0,0 +1,73 @@ +/* eslint-disable @typescript-eslint/unbound-method */ +import type { MigrationContext, ReversibleMigration } from '../migration-types'; + +/** + * Create `workflow_statistics_delta` to store increments to workflow statistics + * that we will later fold into the `workflow_statistics`. This prevents hot-row + * contention on the `workflow_statistics` table in high-throughput scenarios. + * + * This table is a high-churn buffer (insert per execution, delete on fold), so + * its autovacuum is tuned to reclaim the fold's dead tuples aggressively + * instead of letting the table and its PK index bloat. + * + * - `scale_factor = 0` removes the size-proportional part of the vacuum trigger, + * so dead-tuple cleanup does not back off as the table grows during a backlog. + * - `threshold = 1000` fires a vacuum at a flat 1000 dead tuples regardless of + * table size, keeping it responsive. + * - `cost_delay = 0` drops autovacuum's throttling pauses so it runs at full + * speed and keeps up with the delete rate. + */ +export class CreateWorkflowStatisticsDeltaTable1784000000043 implements ReversibleMigration { + async up({ escape, runQuery }: MigrationContext) { + const delta = escape.tableName('workflow_statistics_delta'); + const id = escape.columnName('id'); + const workflowId = escape.columnName('workflowId'); + const name = escape.columnName('name'); + const rootCountDelta = escape.columnName('rootCountDelta'); + const createdAt = escape.columnName('createdAt'); + const workflowName = escape.columnName('workflowName'); + + await runQuery( + `CREATE TABLE ${delta} ( + ${id} BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + ${workflowId} VARCHAR(36) NOT NULL, + ${name} VARCHAR(128) NOT NULL, + ${rootCountDelta} SMALLINT NOT NULL, + ${createdAt} TIMESTAMP(3) WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP, + ${workflowName} VARCHAR(128) + ) WITH ( + autovacuum_vacuum_scale_factor = 0.0, + autovacuum_vacuum_threshold = 1000, + autovacuum_vacuum_cost_delay = 0 + )`, + ); + } + + async down({ escape, runQuery }: MigrationContext) { + const stats = escape.tableName('workflow_statistics'); + const statsDelta = escape.tableName('workflow_statistics_delta'); + const workflowId = escape.columnName('workflowId'); + const name = escape.columnName('name'); + const count = escape.columnName('count'); + const rootCount = escape.columnName('rootCount'); + const rootCountDelta = escape.columnName('rootCountDelta'); + const latestEvent = escape.columnName('latestEvent'); + const createdAt = escape.columnName('createdAt'); + const workflowName = escape.columnName('workflowName'); + + await runQuery( + `INSERT INTO ${stats} (${workflowId}, ${name}, ${count}, ${rootCount}, ${latestEvent}, ${workflowName}) + SELECT ${workflowId}, ${name}, COUNT(*), SUM(${rootCountDelta}), MAX(${createdAt}), + (ARRAY_AGG(${workflowName} ORDER BY ${createdAt} DESC NULLS LAST))[1] + FROM ${statsDelta} + GROUP BY ${workflowId}, ${name} + ON CONFLICT (${name}, ${workflowId}) DO UPDATE SET + ${count} = ${stats}.${count} + EXCLUDED.${count}, + ${rootCount} = ${stats}.${rootCount} + EXCLUDED.${rootCount}, + ${latestEvent} = GREATEST(${stats}.${latestEvent}, EXCLUDED.${latestEvent}), + ${workflowName} = EXCLUDED.${workflowName}`, + ); + + await runQuery(`DROP TABLE ${statsDelta}`); + } +} diff --git a/packages/@n8n/db/src/migrations/postgresdb/index.ts b/packages/@n8n/db/src/migrations/postgresdb/index.ts index 70d38024a59..35b495a5eb7 100644 --- a/packages/@n8n/db/src/migrations/postgresdb/index.ts +++ b/packages/@n8n/db/src/migrations/postgresdb/index.ts @@ -53,6 +53,7 @@ import { ExpandVariablesValueColumnToText1777420800000 } from './1777420800000-E import { LimitWorkflowVersionTriggerToContent1784000000003 } from './1784000000003-LimitWorkflowVersionTriggerToContent'; import { AddProjectIdToInstanceAiThread1784000000028 } from './1784000000028-AddProjectIdToInstanceAiThread'; import { AddExecutionEntityWorkflowStatusIndex1784000000031 } from './1784000000031-AddExecutionEntityWorkflowStatusIndex'; +import { CreateWorkflowStatisticsDeltaTable1784000000043 } from './1784000000043-CreateWorkflowStatisticsDeltaTable'; import { CreateLdapEntities1674509946020 } from '../common/1674509946020-CreateLdapEntities'; import { PurgeInvalidWorkflowConnections1675940580449 } from '../common/1675940580449-PurgeInvalidWorkflowConnections'; import { RemoveResetPasswordColumns1690000000030 } from '../common/1690000000030-RemoveResetPasswordColumns'; @@ -437,4 +438,5 @@ export const postgresMigrations: Migration[] = [ CreateWorkflowPublicationTriggerStatusTable1784000000040, AddUsedPrivateCredentialsToExecutionEntity1784000000041, CreateSchedulerTables1784000000042, + CreateWorkflowStatisticsDeltaTable1784000000043, ]; diff --git a/packages/@n8n/db/src/repositories/workflow-statistics.repository.ts b/packages/@n8n/db/src/repositories/workflow-statistics.repository.ts index 275888ca935..eceee5f9209 100644 --- a/packages/@n8n/db/src/repositories/workflow-statistics.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow-statistics.repository.ts @@ -1,8 +1,8 @@ import { GlobalConfig } from '@n8n/config'; import { Service } from '@n8n/di'; import { PROJECT_OWNER_ROLE_SLUG } from '@n8n/permissions'; -import { DataSource, QueryFailedError, Repository } from '@n8n/typeorm'; -import assert from 'node:assert'; +import { DataSource, type EntityManager, QueryFailedError, Repository } from '@n8n/typeorm'; +import { UnexpectedError } from 'n8n-workflow'; import { ProjectRelation, @@ -17,6 +17,27 @@ import { StatisticsNames } from '../entities/types-db'; type StatisticsInsertResult = 'insert' | 'failed' | 'alreadyExists'; type StatisticsUpsertResult = StatisticsInsertResult | 'update'; +/** + * A counter row the fold just created for the first time, i.e. a workflow's first-ever + * production success or failure. The rollup uses these to fire one-time milestone events. + */ +export type FirstOccurrenceRow = { + name: StatisticsNames; + workflowId: string; + workflowName: string | null; + firstEventMs: number; +}; + +/** One row returned by the fold query: the batch total (`delta_rows`) plus per-counter fold results. */ +type FoldRow = { + delta_rows: number; // `::int` cast in the query -> pg returns int4 as a JS number + workflowId: string; + name: StatisticsNames; + inserted: boolean; + workflowName: string | null; + firstEvent: Date; // timestamptz -> pg driver parses to a JS Date +}; + @Service() export class WorkflowStatisticsRepository extends Repository { constructor( @@ -65,58 +86,129 @@ export class WorkflowStatisticsRepository extends Repository isRootExecution: boolean, workflowName?: string, ): Promise { - const dbType = this.globalConfig.database.type; + if (this.globalConfig.database.type !== 'sqlite') { + throw new UnexpectedError( + 'upsertWorkflowStatistics is SQLite-only; use appendIncrement + rollup on Postgres', + ); + } + const escapedTableName = this.manager.connection.driver.escape(this.metadata.tableName); try { const rootCountIncrement = isRootExecution ? 1 : 0; - if (dbType === 'sqlite') { - await this.query( - `INSERT INTO ${escapedTableName} ("count", "rootCount", "name", "workflowId", "workflowName", "latestEvent") - VALUES (1, ?, ?, ?, ?, CURRENT_TIMESTAMP) - ON CONFLICT (workflowId, name) - DO UPDATE SET - count = count + 1, - rootCount = rootCount + ?, - workflowName = excluded.workflowName, - latestEvent = CURRENT_TIMESTAMP`, - [rootCountIncrement, eventName, workflowId, workflowName ?? null, rootCountIncrement], - ); + await this.query( + `INSERT INTO ${escapedTableName} ("count", "rootCount", "name", "workflowId", "workflowName", "latestEvent") + VALUES (1, ?, ?, ?, ?, CURRENT_TIMESTAMP) + ON CONFLICT (workflowId, name) + DO UPDATE SET + count = count + 1, + rootCount = rootCount + ?, + workflowName = excluded.workflowName, + latestEvent = CURRENT_TIMESTAMP`, + [rootCountIncrement, eventName, workflowId, workflowName ?? null, rootCountIncrement], + ); - // SQLite does not offer a reliable way to know whether or not an insert or update happened. - // We'll use a naive approach in this case. Query again after and it might cause us to miss the - // first production execution sometimes due to concurrency, but it's the only way. - const counter = await this.findOne({ - select: ['count'], - where: { name: eventName, workflowId }, - }); + // SQLite does not offer a reliable way to know whether or not an insert or update happened. + // We'll use a naive approach in this case. Query again after and it might cause us to miss the + // first production execution sometimes due to concurrency, but it's the only way. + const counter = await this.findOne({ + select: ['count'], + where: { name: eventName, workflowId }, + }); - return (counter?.count ?? 0) > 1 ? 'update' : counter?.count === 1 ? 'insert' : 'failed'; - } else if (dbType === 'postgresdb') { - const queryResult = (await this.query( - `INSERT INTO ${escapedTableName} ("count", "rootCount", "name", "workflowId", "workflowName", "latestEvent") - VALUES (1, $1, $2, $3, $4, CURRENT_TIMESTAMP) - ON CONFLICT ("name", "workflowId") - DO UPDATE SET - "count" = ${escapedTableName}."count" + 1, - "rootCount" = ${escapedTableName}."rootCount" + $5, - "workflowName" = $4, - "latestEvent" = CURRENT_TIMESTAMP - RETURNING *;`, - [rootCountIncrement, eventName, workflowId, workflowName ?? null, rootCountIncrement], - )) as Array<{ count: string | number }>; - - return Number(queryResult[0].count) === 1 ? 'insert' : 'update'; - } - - assert.fail('Unknown database type'); + return (counter?.count ?? 0) > 1 ? 'update' : counter?.count === 1 ? 'insert' : 'failed'; } catch (error) { - console.log('error', error); if (error instanceof QueryFailedError) return 'failed'; throw error; } } + private deltaTableName() { + const { tablePrefix } = this.globalConfig.database; + return this.manager.connection.driver.escape(`${tablePrefix}workflow_statistics_delta`); + } + + async appendIncrement( + eventName: StatisticsNames, + workflowId: string, + isRootExecution: boolean, + workflowName?: string, + ): Promise { + const table = this.deltaTableName(); + await this.query( + `INSERT INTO ${table} ("workflowId", "name", "rootCountDelta", "workflowName") VALUES ($1, $2, $3, $4)`, + [workflowId, eventName, isRootExecution ? 1 : 0, workflowName ?? null], + ); + } + + async rollupIncrements( + manager: EntityManager, + batchSize: number, + ): Promise<{ increments: number; firstOccurrences: FirstOccurrenceRow[] }> { + const stats = this.manager.connection.driver.escape(this.metadata.tableName); + const delta = this.deltaTableName(); + + /** + * Query walkthrough + * + * Fold a batch of increments into the target in one atomic statement, so concurrent appends and crashes + * cannot double-count or drop rows. Whatever was not claimed-and-folded stays for next tick. + * + * - `batch` Claim the oldest increments (DELETE + RETURNING, so they cannot be folded twice). + * - `agg` Sum the claimed rows per (workflowId, name) into the totals to add. + * - `upsert` Add the totals into the counter, creating the row if new. `"xmax" = 0` flags rows it created. + * - `SELECT` Return how many increments processed, plus the created (first-ever) rows for milestones. + */ + const rows = await manager.query( + `WITH batch AS ( + DELETE FROM ${delta} + WHERE id IN (SELECT id FROM ${delta} ORDER BY id LIMIT $1) + RETURNING "workflowId", "name", "rootCountDelta", "createdAt", "workflowName" + ), + agg AS ( + SELECT "workflowId", "name", + COUNT(*) AS "countIncrement", + SUM("rootCountDelta") AS "rootCountIncrement", + MAX("createdAt") AS "latestEvent", + MIN("createdAt") AS "firstEvent", + (ARRAY_AGG("workflowName" ORDER BY "createdAt" DESC NULLS LAST))[1] AS "workflowName" + FROM batch GROUP BY "workflowId", "name" + ), + upsert AS ( + INSERT INTO ${stats} ("workflowId", "name", "count", "rootCount", "latestEvent", "workflowName") + SELECT "workflowId", "name", "countIncrement", "rootCountIncrement", "latestEvent", "workflowName" FROM agg + ON CONFLICT ("name", "workflowId") DO UPDATE SET + "count" = ${stats}."count" + EXCLUDED."count", + "rootCount" = ${stats}."rootCount" + EXCLUDED."rootCount", + "latestEvent" = GREATEST(${stats}."latestEvent", EXCLUDED."latestEvent"), + "workflowName" = EXCLUDED."workflowName" + RETURNING "workflowId", "name", ("xmax" = 0) AS inserted + ) + SELECT (SELECT COUNT(*) FROM batch)::int AS delta_rows, + u."workflowId", u."name", u.inserted, a."workflowName", a."firstEvent" + FROM upsert u JOIN agg a ON a."workflowId" = u."workflowId" AND a."name" = u."name"`, + [batchSize], + ); + + if (rows.length === 0) return { increments: 0, firstOccurrences: [] }; + + const firstOccurrences: FirstOccurrenceRow[] = rows + .filter( + (r) => + r.inserted && + (r.name === StatisticsNames.productionSuccess || + r.name === StatisticsNames.productionError), + ) + .map((r) => ({ + name: r.name, + workflowId: r.workflowId, + workflowName: r.workflowName, + firstEventMs: r.firstEvent.getTime(), + })); + + return { increments: rows[0].delta_rows, firstOccurrences }; + } + async queryNumWorkflowsUserHasWithFiveOrMoreProdExecs(userId: User['id']): Promise { const result = await this.createQueryBuilder('ws') .select('COUNT(DISTINCT ws.workflowId)', 'count') diff --git a/packages/@n8n/db/src/services/db-lock.service.ts b/packages/@n8n/db/src/services/db-lock.service.ts index 84c0f07726a..4ab630af880 100644 --- a/packages/@n8n/db/src/services/db-lock.service.ts +++ b/packages/@n8n/db/src/services/db-lock.service.ts @@ -12,6 +12,7 @@ import { OperationalError } from 'n8n-workflow'; export const enum DbLock { AUTH_ROLES_SYNC = 1001, TRUSTED_KEY_REFRESH = 1002, + WORKFLOW_STATISTICS_ROLLUP = 1003, /** Reserved for integration tests — never use in production code */ TEST = 9999, } diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index 5f630662d1e..dee6dd8ebc6 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -41,6 +41,7 @@ import { JwtService } from '@/services/jwt.service'; import { OwnershipService } from '@/services/ownership.service'; import { ExecutionsPruningService } from '@/services/pruning/executions-pruning.service'; import { WorkflowHistoryCompactionService } from '@/services/pruning/workflow-history-compaction.service'; +import { WorkflowStatisticsRollupService } from '@/services/workflow-statistics-rollup.service'; import { UrlService } from '@/services/url.service'; import { WaitTracker } from '@/wait-tracker'; import { WorkflowRunner } from '@/workflow-runner'; @@ -398,6 +399,7 @@ export class Start extends BaseCommand> { Container.get(ExecutionsPruningService).init(); Container.get(WorkflowHistoryCompactionService).init(); + Container.get(WorkflowStatisticsRollupService).init(); Container.get(N8NCheckpointStorage).init(); if (this.globalConfig.executions.mode === 'regular') { diff --git a/packages/cli/src/databases/repositories/__tests__/workflow-statistics.integration.test.ts b/packages/cli/src/databases/repositories/__tests__/workflow-statistics.integration.test.ts index e7b2371e843..748af9155ad 100644 --- a/packages/cli/src/databases/repositories/__tests__/workflow-statistics.integration.test.ts +++ b/packages/cli/src/databases/repositories/__tests__/workflow-statistics.integration.test.ts @@ -6,6 +6,11 @@ import { mock, mockClear } from 'vitest-mock-extended'; import { mockEntityManager } from '@test/mocking'; +// `upsertWorkflowStatistics` is SQLite-only; on Postgres, recording goes through appendIncrement + rollup. +const runOnSqlite = (process.env.DB_TYPE ?? 'sqlite') === 'sqlite'; +// eslint-disable-next-line n8n-local-rules/no-skipped-tests -- SQLite-only method +const describeSqlite = runOnSqlite ? describe : describe.skip; + describe('insertWorkflowStatistics', () => { const entityManager = mockEntityManager(WorkflowStatistics); const workflowStatisticsRepository = Container.get(WorkflowStatisticsRepository); @@ -52,7 +57,7 @@ describe('insertWorkflowStatistics', () => { }); }); -describe('upsertWorkflowStatistics', () => { +describeSqlite('upsertWorkflowStatistics', () => { let repository: WorkflowStatisticsRepository; beforeAll(async () => { Container.reset(); 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 new file mode 100644 index 00000000000..6ca0a361fa2 --- /dev/null +++ b/packages/cli/src/services/__tests__/workflow-statistics-rollup.service.test.ts @@ -0,0 +1,279 @@ +import type { Logger } from '@n8n/backend-common'; +import type { DatabaseConfig } from '@n8n/config'; +import type { DbConnection, DbLockService, WorkflowStatisticsRepository } from '@n8n/db'; +import { StatisticsNames } from '@n8n/db'; +import { mock } from 'vitest-mock-extended'; +import type { ErrorReporter, InstanceSettings } from 'n8n-core'; +import { OperationalError } from 'n8n-workflow'; + +import type { WorkflowStatisticsService } from '../workflow-statistics.service'; +import { WorkflowStatisticsRollupService } from '../workflow-statistics-rollup.service'; + +type RollupResult = Awaited>; + +describe('WorkflowStatisticsRollupService', () => { + const dbConnection = mock({ connectionState: { migrated: true } }); + + const makeService = (opts: { + isLeader?: boolean; + instanceType?: 'main' | 'worker'; + dbType?: 'postgresdb' | 'sqlite'; + }) => { + const instanceSettings = mock({ + isLeader: opts.isLeader ?? true, + instanceType: opts.instanceType ?? 'main', + instanceRole: 'leader', + }); + const databaseConfig = mock({ type: opts.dbType ?? 'postgresdb' }); + const errorReporter = mock(); + const dbLockService = mock(); + const repository = mock(); + const statisticsService = mock(); + const logger = mock(); // the scoped logger the service actually logs to + const service = new WorkflowStatisticsRollupService( + mock({ scoped: vi.fn().mockReturnValue(logger) }), + errorReporter, + instanceSettings, + dbConnection, + databaseConfig, + dbLockService, + repository, + statisticsService, + ); + return { service, logger, errorReporter, dbLockService, repository, statisticsService }; + }; + + /** Invoke the private rollup tick directly. */ + const rollup = async (service: WorkflowStatisticsRollupService) => + await (service as unknown as { rollup: () => Promise }).rollup(); + + describe('shouldRun', () => { + it('is true for a leader main on Postgres', () => { + const { service } = makeService({ + isLeader: true, + instanceType: 'main', + dbType: 'postgresdb', + }); + expect(service.shouldRun).toBe(true); + }); + + it('is false on SQLite', () => { + const { service } = makeService({ isLeader: true, instanceType: 'main', dbType: 'sqlite' }); + expect(service.shouldRun).toBe(false); + }); + + it('is false for a follower', () => { + const { service } = makeService({ isLeader: false }); + expect(service.shouldRun).toBe(false); + }); + + it('is false for a non-main instance', () => { + const { service } = makeService({ instanceType: 'worker' }); + expect(service.shouldRun).toBe(false); + }); + }); + + describe('init', () => { + it('starts the rollup on a leader main (Postgres)', () => { + const { service } = makeService({ isLeader: true }); + const startSpy = vi.spyOn(service, 'start').mockImplementation(() => {}); + service.init(); + expect(startSpy).toHaveBeenCalled(); + }); + + it('does not start on a follower', () => { + const { service } = makeService({ isLeader: false }); + const startSpy = vi.spyOn(service, 'start').mockImplementation(() => {}); + service.init(); + expect(startSpy).not.toHaveBeenCalled(); + }); + + it('does not run after shutdown', async () => { + const { service } = makeService({ isLeader: true }); + expect(service.shouldRun).toBe(true); + await service.shutdown(); + expect(service.shouldRun).toBe(false); + }); + }); + + describe('start', () => { + const isRunning = (service: WorkflowStatisticsRollupService) => + (service as unknown as { timeout: NodeJS.Timeout | undefined }).timeout !== undefined; + + it('does not schedule on a takeover when it should not run', () => { + const { service } = makeService({ dbType: 'sqlite' }); + service.start(); + expect(isRunning(service)).toBe(false); + }); + }); + + describe('lock contention', () => { + const lockHeld = () => new OperationalError('lock held'); + + it('warns with counts once skips reach the threshold', async () => { + const { service, dbLockService, logger } = makeService({}); + dbLockService.tryWithLock.mockRejectedValue(lockHeld()); + + for (let i = 0; i < 5; i++) await rollup(service); + + expect(logger.warn).toHaveBeenCalledTimes(1); + expect(logger.warn).toHaveBeenCalledWith(expect.any(String), { + consecutiveLockSkips: 5, + totalLockSkips: 5, + }); + }); + + it('resets the consecutive count on a successful fold but keeps the total', async () => { + const { service, dbLockService, logger } = makeService({}); + const empty: RollupResult = { increments: 0, firstOccurrences: [] }; + + for (let i = 0; i < 4; i++) dbLockService.tryWithLock.mockRejectedValueOnce(lockHeld()); + dbLockService.tryWithLock.mockResolvedValueOnce(empty); + for (let i = 0; i < 5; i++) await rollup(service); + expect(logger.warn).not.toHaveBeenCalled(); // streak broken at 4 + + for (let i = 0; i < 5; i++) dbLockService.tryWithLock.mockRejectedValueOnce(lockHeld()); + for (let i = 0; i < 5; i++) await rollup(service); + + expect(logger.warn).toHaveBeenCalledTimes(1); + expect(logger.warn).toHaveBeenCalledWith(expect.any(String), { + consecutiveLockSkips: 5, + totalLockSkips: 9, + }); + }); + }); + + describe('shutdown', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('resolves immediately when no tick is in flight', async () => { + const { service } = makeService({}); + await expect(service.shutdown()).resolves.toBeUndefined(); + }); + + it('awaits the in-flight tick (fold and milestones) before resolving, then stops', async () => { + const { service, dbLockService, statisticsService } = makeService({}); + let resolveFold!: (result: RollupResult) => void; + dbLockService.tryWithLock.mockImplementation( + async () => await new Promise((resolve) => (resolveFold = resolve)), + ); + statisticsService.emitFirstOccurrenceEvent.mockResolvedValue(); + + service.start(); + await vi.advanceTimersByTimeAsync(0); // fire the first tick; fold now in flight + + let shutdownSettled = false; + const shutdownPromise = service.shutdown().then(() => (shutdownSettled = true)); + await vi.advanceTimersByTimeAsync(0); // flush microtasks + expect(shutdownSettled).toBe(false); // still awaiting the in-flight fold + + resolveFold({ + increments: 1, + firstOccurrences: [ + { + name: StatisticsNames.productionSuccess, + workflowId: 'wf-1', + workflowName: 'A', + firstEventMs: 1, + }, + ], + }); + await shutdownPromise; + + // The milestone was emitted before shutdown resolved. + expect(statisticsService.emitFirstOccurrenceEvent).toHaveBeenCalledTimes(1); + + // No further tick is scheduled after shutdown. + await vi.advanceTimersByTimeAsync(60_000); + expect(dbLockService.tryWithLock).toHaveBeenCalledTimes(1); + }); + }); + + describe('rollup', () => { + it('skips the tick (returns 0, no milestones) when the advisory lock is held by another instance', async () => { + const { service, dbLockService, statisticsService } = makeService({}); + dbLockService.tryWithLock.mockRejectedValue(new OperationalError('lock held')); + + const folded = await rollup(service); + + expect(folded).toBe(0); + expect(statisticsService.emitFirstOccurrenceEvent).not.toHaveBeenCalled(); + }); + + it('rethrows a non-lock error so the scheduler logs and retries the tick', async () => { + const { service, dbLockService, statisticsService } = makeService({}); + dbLockService.tryWithLock.mockRejectedValue(new Error('connection reset')); + + await expect(rollup(service)).rejects.toThrow('connection reset'); + expect(statisticsService.emitFirstOccurrenceEvent).not.toHaveBeenCalled(); + }); + + it('fires a milestone for each first-occurrence row and returns the increment count', async () => { + const { service, dbLockService, statisticsService } = makeService({}); + const result: RollupResult = { + increments: 1234, + firstOccurrences: [ + { + name: StatisticsNames.productionSuccess, + workflowId: 'wf-1', + workflowName: 'A', + firstEventMs: 1, + }, + { + name: StatisticsNames.productionError, + workflowId: 'wf-2', + workflowName: 'B', + firstEventMs: 2, + }, + ], + }; + dbLockService.tryWithLock.mockResolvedValue(result); + + const folded = await rollup(service); + + expect(folded).toBe(1234); + expect(statisticsService.emitFirstOccurrenceEvent).toHaveBeenCalledTimes(2); + expect(statisticsService.emitFirstOccurrenceEvent).toHaveBeenCalledWith( + StatisticsNames.productionSuccess, + 'wf-1', + 'A', + 1, + ); + }); + + it('isolates milestone failures: one throwing row does not stop the others or fail the tick', async () => { + const { service, dbLockService, statisticsService } = makeService({}); + const result: RollupResult = { + increments: 2, + firstOccurrences: [ + { + name: StatisticsNames.productionSuccess, + workflowId: 'deleted-wf', + workflowName: null, + firstEventMs: 1, + }, + { + name: StatisticsNames.productionSuccess, + workflowId: 'wf-2', + workflowName: 'B', + firstEventMs: 2, + }, + ], + }; + dbLockService.tryWithLock.mockResolvedValue(result); + statisticsService.emitFirstOccurrenceEvent + .mockRejectedValueOnce(new Error('workflow/project deleted during rollup lag')) + .mockResolvedValueOnce(); + + // Must not throw, and must still return the folded count. + await expect(rollup(service)).resolves.toBe(2); + expect(statisticsService.emitFirstOccurrenceEvent).toHaveBeenCalledTimes(2); + }); + }); +}); 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 bfc9c10048c..2d720dd1740 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 @@ -6,13 +6,13 @@ import { testDb, mockInstance, } from '@n8n/backend-test-utils'; -import { GlobalConfig } from '@n8n/config'; +import { DatabaseConfig, GlobalConfig } from '@n8n/config'; import type { IWorkflowDb, Project, WorkflowEntity, WorkflowRepository, User } from '@n8n/db'; -import { SettingsRepository, WorkflowStatisticsRepository } from '@n8n/db'; +import { SettingsRepository, StatisticsNames, WorkflowStatisticsRepository } from '@n8n/db'; import { Container } from '@n8n/di'; import { QueryFailedError, - type DataSource, + DataSource, type EntityManager, type EntityMetadata, } from '@n8n/typeorm'; @@ -40,24 +40,59 @@ describe('WorkflowStatisticsService', () => { let user: User; let personalProject: Project; let workflow: IWorkflowDb & WorkflowEntity; + let dataSource: DataSource; + let isPostgres: boolean; beforeAll(async () => { await testDb.init(); workflowStatisticsService = Container.get(WorkflowStatisticsService); workflowStatisticsRepository = Container.get(WorkflowStatisticsRepository); userService = Container.get(UserService); + dataSource = Container.get(DataSource); + isPostgres = Container.get(GlobalConfig).database.type === 'postgresdb'; user = await createUser(); personalProject = await getPersonalProject(user); workflow = await createWorkflow({}, user); }); + /** + * On Postgres, `workflowExecutionCompleted` appends to the delta table and the rollup folds it + * out of band. These tests assert the materialized counter + milestone, so we drive the fold + * here (the rollup's work, minus the leader/lock/timer) to mimic the synchronous behavior the + * SQLite path still has. On SQLite this is a no-op (the upsert path already materialized). + */ + const flushStats = async (service: WorkflowStatisticsService) => { + if (!isPostgres) return; + const { firstOccurrences } = await workflowStatisticsRepository.rollupIncrements( + dataSource.manager, + 10_000, + ); + for (const occ of firstOccurrences) { + await service.emitFirstOccurrenceEvent( + occ.name, + occ.workflowId, + occ.workflowName, + occ.firstEventMs, + ); + } + }; + + const completeAndFlush = async ( + service: WorkflowStatisticsService, + workflowData: IWorkflowDb & WorkflowEntity, + runData: IRun, + ) => { + await service.workflowExecutionCompleted(workflowData, runData); + await flushStats(service); + }; + afterAll(async () => { await testDb.terminate(); }); beforeEach(async () => { vi.restoreAllMocks(); - await testDb.truncate(['WorkflowStatistics']); + await testDb.truncate(['WorkflowStatistics', 'WorkflowStatisticsDelta']); // Clear first production failure setting const settingsRepository = Container.get(SettingsRepository); await settingsRepository.delete({ key: 'instance.firstProductionFailure' }); @@ -77,8 +112,8 @@ describe('WorkflowStatisticsService', () => { }; // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT const statistics = await workflowStatisticsRepository.find(); @@ -110,8 +145,8 @@ describe('WorkflowStatisticsService', () => { }; // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT const statistics = await workflowStatisticsRepository.find(); @@ -141,7 +176,7 @@ describe('WorkflowStatisticsService', () => { storedAt: 'db', }; - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); const statistics = await workflowStatisticsRepository.find(); expect(statistics).toHaveLength(0); @@ -162,8 +197,8 @@ describe('WorkflowStatisticsService', () => { }; // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT const statistics = await workflowStatisticsRepository.find(); @@ -193,7 +228,7 @@ describe('WorkflowStatisticsService', () => { }; // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT const statistics = await workflowStatisticsRepository.find(); @@ -215,14 +250,16 @@ describe('WorkflowStatisticsService', () => { const updateSettingsSpy = vi.spyOn(userService, 'updateSettings'); // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT expect(updateSettingsSpy).toHaveBeenCalledTimes(1); expect(updateSettingsSpy).toHaveBeenCalledWith(user.id, { firstSuccessfulWorkflowId: workflow.id, userActivated: true, - userActivatedAt: runData.startedAt.getTime(), + // On Postgres the milestone fires from the fold using the delta's `firstEvent` + // (~completion time), not `runData.startedAt` — an accepted best-effort drift. + userActivatedAt: isPostgres ? expect.any(Number) : runData.startedAt.getTime(), }); expect(emitSpy).toHaveBeenCalledTimes(1); expect(emitSpy).toHaveBeenCalledWith('first-production-workflow-succeeded', { @@ -246,7 +283,7 @@ describe('WorkflowStatisticsService', () => { const updateSettingsSpy = vi.spyOn(userService, 'updateSettings'); // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT expect(updateSettingsSpy).not.toHaveBeenCalled(); @@ -266,12 +303,12 @@ describe('WorkflowStatisticsService', () => { startedAt: new Date(), storedAt: 'db', }; - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); const updateSettingsSpy = vi.spyOn(Container.get(UserService), 'updateSettings'); // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT expect(updateSettingsSpy).not.toHaveBeenCalled(); @@ -305,11 +342,12 @@ describe('WorkflowStatisticsService', () => { Container.get(EventService), settingsRepository, workflowRepositoryNoErrorWorkflows, + Container.get(DatabaseConfig), ); const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); // ACT - await statisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(statisticsService, workflow, runData); // ASSERT expect(emitSpy).toHaveBeenCalledWith('instance-first-production-workflow-failed', { @@ -347,14 +385,15 @@ describe('WorkflowStatisticsService', () => { Container.get(EventService), settingsRepository, workflowRepositoryNoErrorWorkflows, + Container.get(DatabaseConfig), ); // First failure - this will set the instance.firstProductionFailure setting - await statisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(statisticsService, workflow, runData); const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); // ACT - Second failure - await statisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(statisticsService, workflow, runData); // ASSERT expect(emitSpy).not.toHaveBeenCalled(); @@ -387,11 +426,12 @@ describe('WorkflowStatisticsService', () => { Container.get(EventService), settingsRepository, workflowRepositoryWithErrorWorkflows, + Container.get(DatabaseConfig), ); const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); // ACT - await statisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(statisticsService, workflow, runData); // ASSERT expect(emitSpy).not.toHaveBeenCalledWith( @@ -413,7 +453,7 @@ describe('WorkflowStatisticsService', () => { const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT expect(emitSpy).not.toHaveBeenCalledWith( @@ -435,7 +475,7 @@ describe('WorkflowStatisticsService', () => { const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); // ACT - await workflowStatisticsService.workflowExecutionCompleted(workflow, runData); + await completeAndFlush(workflowStatisticsService, workflow, runData); // ASSERT expect(emitSpy).not.toHaveBeenCalledWith( @@ -462,7 +502,7 @@ describe('WorkflowStatisticsService', () => { const updateSettingsSpy = vi.spyOn(userService, 'updateSettings'); // ACT - await workflowStatisticsService.workflowExecutionCompleted(teamWorkflow, runData); + await completeAndFlush(workflowStatisticsService, teamWorkflow, runData); // ASSERT expect(updateSettingsSpy).not.toHaveBeenCalled(); @@ -504,6 +544,7 @@ describe('WorkflowStatisticsService', () => { Container.get(EventService), settingsRepository, workflowRepositoryNoErrorWorkflows, + Container.get(DatabaseConfig), ); const emitSpy = vi.spyOn(Container.get(EventService), 'emit'); @@ -511,7 +552,7 @@ describe('WorkflowStatisticsService', () => { const instanceOwner = await ownershipService.getInstanceOwner(); // ACT - await statisticsService.workflowExecutionCompleted(teamWorkflow, runData); + await completeAndFlush(statisticsService, teamWorkflow, runData); // ASSERT // For team projects, it should fall back to instance owner @@ -522,6 +563,239 @@ describe('WorkflowStatisticsService', () => { userId: instanceOwner.id, }); }); + + // 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. + const deltaTable = () => + `${Container.get(GlobalConfig).database.tablePrefix}workflow_statistics_delta`; + + const append = async (isRoot: boolean) => + await workflowStatisticsRepository.appendIncrement( + StatisticsNames.productionSuccess, + workflow.id, + isRoot, + workflow.name, + ); + + const appendFor = async (name: StatisticsNames, workflowId: string, isRoot: boolean) => + await workflowStatisticsRepository.appendIncrement(name, workflowId, isRoot, 'wf'); + + const countersByKey = async () => { + const rows = await workflowStatisticsRepository.find(); + return new Map(rows.map((r) => [`${r.workflowId}|${r.name}`, r])); + }; + + test('folds many appended deltas into one counter row with exact totals', async () => { + if (!isPostgres) return; + + // 5 appends: 3 root, 2 non-root + await append(true); + await append(true); + await append(true); + await append(false); + await append(false); + + const { increments, firstOccurrences } = + await workflowStatisticsRepository.rollupIncrements(dataSource.manager, 10_000); + + expect(increments).toBe(5); + expect(firstOccurrences).toHaveLength(1); // the new production_success counter row + expect(firstOccurrences[0]).toMatchObject({ + name: 'production_success', + workflowId: workflow.id, + }); + + const [counter] = await workflowStatisticsRepository.find(); + expect(counter).toMatchObject({ count: 5, rootCount: 3, name: 'production_success' }); + + const remaining = await dataSource.query(`SELECT COUNT(*)::int AS c FROM ${deltaTable()}`); + expect(remaining[0].c).toBe(0); // delta drained + }); + + test('drains a backlog larger than the batch size across multiple folds (bounded)', async () => { + if (!isPostgres) return; + + for (let i = 0; i < 5; i++) await append(true); + + // Batch size 2 -> folds of 2, 2, 1, then 0 (drained). + const counts: number[] = []; + let folded: number; + do { + ({ increments: folded } = await workflowStatisticsRepository.rollupIncrements( + dataSource.manager, + 2, + )); + counts.push(folded); + } while (folded > 0); + + expect(counts).toEqual([2, 2, 1, 0]); + + const [counter] = await workflowStatisticsRepository.find(); + expect(counter).toMatchObject({ count: 5, rootCount: 5 }); + }); + + test('does not report a first occurrence when the counter row already exists', async () => { + if (!isPostgres) return; + + // First occurrence: creates the counter row. + await append(true); + const first = await workflowStatisticsRepository.rollupIncrements( + dataSource.manager, + 10_000, + ); + expect(first.firstOccurrences).toHaveLength(1); + + // Subsequent fold updates the existing row -> no first-occurrence -> no milestone. + await append(true); + const second = await workflowStatisticsRepository.rollupIncrements( + dataSource.manager, + 10_000, + ); + expect(second.increments).toBe(1); + expect(second.firstOccurrences).toHaveLength(0); + + const [counter] = await workflowStatisticsRepository.find(); + expect(counter).toMatchObject({ count: 2 }); + }); + + test('folds several (workflow, name) groups in one batch, each with its own totals', async () => { + if (!isPostgres) return; + + const workflow2 = await createWorkflow({}, user); + + // Three groups folded together: exercises GROUP BY + the upsert/agg join. + await appendFor(StatisticsNames.productionSuccess, workflow.id, true); + await appendFor(StatisticsNames.productionSuccess, workflow.id, true); + await appendFor(StatisticsNames.productionSuccess, workflow.id, false); + await appendFor(StatisticsNames.productionError, workflow.id, true); + await appendFor(StatisticsNames.productionSuccess, workflow2.id, true); + await appendFor(StatisticsNames.productionSuccess, workflow2.id, true); + + const { increments, firstOccurrences } = + await workflowStatisticsRepository.rollupIncrements(dataSource.manager, 10_000); + + expect(increments).toBe(6); + + // Every group is new, so each is reported once and attributed to the right (workflow, name). + const occKeys = new Set(firstOccurrences.map((o) => `${o.workflowId}|${o.name}`)); + expect(occKeys).toEqual( + new Set([ + `${workflow.id}|production_success`, + `${workflow.id}|production_error`, + `${workflow2.id}|production_success`, + ]), + ); + + const counters = await countersByKey(); + expect(counters.get(`${workflow.id}|production_success`)).toMatchObject({ + count: 3, + rootCount: 2, + }); + expect(counters.get(`${workflow.id}|production_error`)).toMatchObject({ + count: 1, + rootCount: 1, + }); + expect(counters.get(`${workflow2.id}|production_success`)).toMatchObject({ + count: 2, + rootCount: 2, + }); + }); + + test('reports a first occurrence only for groups whose counter row is new', async () => { + if (!isPostgres) return; + + const workflow2 = await createWorkflow({}, user); + + // Pre-existing counter row for workflow1/success. + await appendFor(StatisticsNames.productionSuccess, workflow.id, true); + await workflowStatisticsRepository.rollupIncrements(dataSource.manager, 10_000); + + // One batch touching the existing group and a brand-new one. + await appendFor(StatisticsNames.productionSuccess, workflow.id, true); + await appendFor(StatisticsNames.productionSuccess, workflow2.id, true); + + const { increments, firstOccurrences } = + await workflowStatisticsRepository.rollupIncrements(dataSource.manager, 10_000); + + expect(increments).toBe(2); + expect(firstOccurrences).toHaveLength(1); + expect(firstOccurrences[0]).toMatchObject({ + workflowId: workflow2.id, + name: 'production_success', + }); + + const counters = await countersByKey(); + expect(counters.get(`${workflow.id}|production_success`)).toMatchObject({ count: 2 }); + expect(counters.get(`${workflow2.id}|production_success`)).toMatchObject({ count: 1 }); + }); + + test('does not regress an existing counter row to an older event time', async () => { + if (!isPostgres) return; + + const future = new Date(Date.now() + 24 * 60 * 60 * 1000); + await workflowStatisticsRepository.insert({ + workflowId: workflow.id, + name: StatisticsNames.productionSuccess, + count: 1, + rootCount: 1, + latestEvent: future, + workflowName: workflow.name, + }); + + // Folding "now" events into a row whose stored timestamp is in the future. + await append(true); + await workflowStatisticsRepository.rollupIncrements(dataSource.manager, 10_000); + + const [counter] = await workflowStatisticsRepository.find(); + expect(counter).toMatchObject({ count: 2, rootCount: 2 }); + // GREATEST keeps the newer stored timestamp, not the older folded one. + expect(counter.latestEvent.getTime()).toBe(future.getTime()); + }); + + test('returns an empty result when there are no deltas', async () => { + if (!isPostgres) return; + + const result = await workflowStatisticsRepository.rollupIncrements( + dataSource.manager, + 10_000, + ); + + expect(result).toEqual({ increments: 0, firstOccurrences: [] }); + }); + + test('carries the earliest event as the first-occurrence time and the latest as the counter time', async () => { + if (!isPostgres) return; + + // Two deltas with explicit, ordered timestamps so MIN/MAX are deterministic. + const earliest = new Date('2030-01-01T00:00:00.000Z'); + const latest = new Date('2030-01-02T00:00:00.000Z'); + await dataSource.query( + `INSERT INTO ${deltaTable()} ("workflowId", "name", "rootCountDelta", "createdAt", "workflowName") + VALUES ($1, $2, 1, $3, 'older'), ($4, $5, 1, $6, 'newer')`, + [ + workflow.id, + StatisticsNames.productionSuccess, + earliest, + workflow.id, + StatisticsNames.productionSuccess, + latest, + ], + ); + + const { firstOccurrences } = await workflowStatisticsRepository.rollupIncrements( + dataSource.manager, + 10_000, + ); + + expect(firstOccurrences).toHaveLength(1); + expect(firstOccurrences[0].firstEventMs).toBe(earliest.getTime()); // MIN(createdAt) + + const [counter] = await workflowStatisticsRepository.find(); + expect(counter.latestEvent.getTime()).toBe(latest.getTime()); // MAX(createdAt) + expect(counter.workflowName).toBe('newer'); // name from the most recent delta + }); + }); }); describe('nodeFetchedData', () => { @@ -540,6 +814,7 @@ describe('WorkflowStatisticsService', () => { ownershipService = mockInstance(OwnershipService); userService = mockInstance(UserService); const globalConfig = Container.get(GlobalConfig); + const databaseConfig = Container.get(DatabaseConfig); entityManager = mock(); const dataSource = mock({ @@ -566,6 +841,7 @@ describe('WorkflowStatisticsService', () => { eventService, settingsRepository, workflowRepository, + databaseConfig, ); globalConfig.diagnostics.enabled = true; globalConfig.deployment.type = 'n8n-testing'; diff --git a/packages/cli/src/services/workflow-statistics-rollup.service.ts b/packages/cli/src/services/workflow-statistics-rollup.service.ts new file mode 100644 index 00000000000..e4874c1ff64 --- /dev/null +++ b/packages/cli/src/services/workflow-statistics-rollup.service.ts @@ -0,0 +1,174 @@ +import { Logger } from '@n8n/backend-common'; +import { DatabaseConfig } from '@n8n/config'; +import { Time } from '@n8n/constants'; +import { DbConnection, DbLock, DbLockService, WorkflowStatisticsRepository } from '@n8n/db'; +import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators'; +import { Service } from '@n8n/di'; +import { ErrorReporter, InstanceSettings } from 'n8n-core'; +import { OperationalError } from 'n8n-workflow'; +import { strict } from 'node:assert'; + +import { WorkflowStatisticsService } from './workflow-statistics.service'; + +type RollupResult = Awaited>; + +const BATCH_SIZE = 5000; + +/** Steady cadence once the backlog is drained. */ +const STEADY_INTERVAL_MS = 5 * Time.seconds.toMilliseconds; + +/** Fast cadence while a batch comes back full, i.e. backlog remaining. */ +const BUSY_INTERVAL_MS = 250; + +/** Consecutive lock skips after which to warn that the lock is persistently held elsewhere. */ +const SKIP_WARN_THRESHOLD = 5; + +/** + * Folds statistics increments in the `workflow_statistics_delta` table + * into the `workflow_statistics` table. + */ +@Service() +export class WorkflowStatisticsRollupService { + private timeout: NodeJS.Timeout | undefined; + + /** Tracks the in-flight tick so `shutdown` can await it. */ + private inflightRollup: Promise | undefined; + + private isShuttingDown = false; + + private consecutiveLockSkips = 0; + + private totalLockSkips = 0; + + constructor( + private readonly logger: Logger, + private readonly errorReporter: ErrorReporter, + private readonly instanceSettings: InstanceSettings, + private readonly dbConnection: DbConnection, + private readonly databaseConfig: DatabaseConfig, + private readonly dbLockService: DbLockService, + private readonly repository: WorkflowStatisticsRepository, + private readonly statisticsService: WorkflowStatisticsService, + ) { + this.logger = this.logger.scoped('workflow-statistics'); + } + + get shouldRun() { + return ( + this.databaseConfig.type === 'postgresdb' && + this.dbConnection.connectionState.migrated && + this.instanceSettings.instanceType === 'main' && + this.instanceSettings.isLeader && + !this.isShuttingDown + ); + } + + init() { + strict(this.instanceSettings.instanceRole !== 'unset', 'Instance role is not set'); + + if (this.shouldRun) this.start(); + } + + @OnLeaderTakeover() + start() { + if (!this.shouldRun || this.timeout !== undefined) return; + + this.scheduleNext(0); + this.logger.debug('Workflow statistics rollup interval started'); + } + + @OnLeaderStepdown() + stop() { + clearTimeout(this.timeout); + this.timeout = undefined; + } + + @OnShutdown() + async shutdown() { + this.isShuttingDown = true; + clearTimeout(this.timeout); + this.timeout = undefined; + await this.inflightRollup; + } + + private scheduleNext(delayMs: number) { + this.timeout = setTimeout(() => { + let nextDelayMs = STEADY_INTERVAL_MS; + this.inflightRollup = this.rollup() + .then((increments) => { + if (increments >= BATCH_SIZE) nextDelayMs = BUSY_INTERVAL_MS; // backlog remaining + }) + .catch((error) => { + this.errorReporter.error(error, { shouldBeLogged: true }); + }) + .finally(() => { + this.inflightRollup = undefined; + if (this.timeout !== undefined) this.scheduleNext(nextDelayMs); + }); + }, delayMs); + } + + /** Fold one batch of increments and fire any resulting milestones. Returns increments folded. */ + private async rollup(): Promise { + const result = await this.foldBatch(); + if (!result) return 0; + + await this.emitMilestones(result.firstOccurrences); + + return result.increments; + } + + /** Fold a batch under an advisory lock. Returns null if another instance holds the lock. */ + private async foldBatch(): Promise { + try { + const result = await this.dbLockService.tryWithLock( + DbLock.WORKFLOW_STATISTICS_ROLLUP, + async (tx) => await this.repository.rollupIncrements(tx, BATCH_SIZE), + ); + this.consecutiveLockSkips = 0; + return result; + } catch (error) { + if (error instanceof OperationalError) { + this.registerLockSkip(); // another instance holds the lock + return null; + } + throw error; + } + } + + /** + * 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). + */ + private registerLockSkip() { + this.consecutiveLockSkips++; + this.totalLockSkips++; + + if (this.consecutiveLockSkips % SKIP_WARN_THRESHOLD !== 0) return; + + this.logger.warn( + 'Workflow statistics rollup repeatedly skipped: lock held by another process', + { + consecutiveLockSkips: this.consecutiveLockSkips, + totalLockSkips: this.totalLockSkips, + }, + ); + } + + /** Fire the one-time milestone event for each workflow whose counter row was just created. */ + private async emitMilestones(firstOccurrences: RollupResult['firstOccurrences']): Promise { + for (const occurrence of firstOccurrences) { + try { + await this.statisticsService.emitFirstOccurrenceEvent( + occurrence.name, + occurrence.workflowId, + occurrence.workflowName, + occurrence.firstEventMs, + ); + } catch (error) { + this.errorReporter.error(error, { shouldBeLogged: true }); + } + } + } +} diff --git a/packages/cli/src/services/workflow-statistics.service.ts b/packages/cli/src/services/workflow-statistics.service.ts index 0478fbcaa1e..1a4f0832fdc 100644 --- a/packages/cli/src/services/workflow-statistics.service.ts +++ b/packages/cli/src/services/workflow-statistics.service.ts @@ -1,4 +1,5 @@ import { Logger, TypedEmitter } from '@n8n/backend-common'; +import { DatabaseConfig } from '@n8n/config'; import { SettingsRepository, StatisticsNames, @@ -6,6 +7,7 @@ import { WorkflowStatisticsRepository, } from '@n8n/db'; import { Service } from '@n8n/di'; +import { ensureError } from '@n8n/utils/errors/ensure-error'; import { isCompletedExecutionStatus, type ExecutionStatus, @@ -93,6 +95,7 @@ export class WorkflowStatisticsService extends TypedEmitter>; + try { - const upsertResult = await this.repository.upsertWorkflowStatistics( + /** + * For performance reasons, in Postgres we append and fold out of band, + * whereas in SQLite we upsert directly. + */ + if (this.databaseConfig.type === 'postgresdb') { + await this.repository.appendIncrement( + statisticsName, + workflowId, + isRoot, + workflowData.name, + ); + return; + } + + upsertResult = await this.repository.upsertWorkflowStatistics( statisticsName, workflowId, - isRootExecutionForRun(runData), + isRoot, workflowData.name, ); - - if (upsertResult !== 'insert') return; - - if (statisticsName === StatisticsNames.productionSuccess) { - await this.emitFirstProductionWorkflowSucceeded(workflowId, runData); - } else if (statisticsName === StatisticsNames.productionError) { - await this.emitInstanceFirstProductionWorkflowFailed(workflowData, workflowId, runData); - } } catch (error) { - this.logger.debug('Unable to fire first workflow telemetry event'); + this.logger.error('Failed to record workflow statistic', { error: ensureError(error) }); + return; + } + + if (upsertResult !== 'insert') return; + + try { + await this.emitFirstOccurrenceEvent( + statisticsName, + workflowId, + workflowData.name ?? null, + runData.startedAt.getTime(), + ); + } catch (error) { + this.logger.debug('Failed to emit workflow statistics milestone', { + error: ensureError(error), + }); + } + } + + /** Emit an event on first production success or first production failure. */ + async emitFirstOccurrenceEvent( + statisticsName: StatisticsNames, + workflowId: string, + workflowName: string | null, + firstEventMs: number, + ): Promise { + if (statisticsName === StatisticsNames.productionSuccess) { + await this.emitFirstProductionWorkflowSucceeded(workflowId, firstEventMs); + return; + } + + if (statisticsName === StatisticsNames.productionError) { + await this.emitInstanceFirstProductionWorkflowFailed( + workflowId, + workflowName ?? '', + firstEventMs, + ); } } private async emitFirstProductionWorkflowSucceeded( workflowId: string, - runData: IRun, + userActivatedAtMs: number, ): Promise { const project = await this.ownershipService.getWorkflowProjectCached(workflowId); let userId: string | null = null; @@ -150,7 +200,7 @@ export class WorkflowStatisticsService extends TypedEmitter { const instanceHadProductionFailure = await this.settingsRepository.findByKey( 'instance.firstProductionFailure', @@ -193,7 +243,7 @@ export class WorkflowStatisticsService extends TypedEmitter { expectedCount: number, timeout = 10000, ): Promise { + const isPostgres = Container.get(GlobalConfig).database.type === 'postgresdb'; const start = Date.now(); while (Date.now() - start < timeout) { + if (isPostgres) { + await workflowStatisticsRepository.rollupIncrements( + workflowStatisticsRepository.manager, + 10_000, + ); + } const stats = await workflowStatisticsRepository.findOne({ where: { workflowId, diff --git a/packages/cli/test/integration/license-metrics.repository.test.ts b/packages/cli/test/integration/license-metrics.repository.test.ts index 4472e3c825e..64bbb681e99 100644 --- a/packages/cli/test/integration/license-metrics.repository.test.ts +++ b/packages/cli/test/integration/license-metrics.repository.test.ts @@ -62,10 +62,9 @@ describe('LicenseMetricsRepository', () => { StatisticsNames.manualError, secondWorkflow.id, ), - workflowStatisticsRepository.upsertWorkflowStatistics( + workflowStatisticsRepository.insertWorkflowStatistics( StatisticsNames.productionSuccess, secondWorkflow.id, - true, ), ]); diff --git a/packages/cli/test/migration/1784000000043-create-workflow-statistics-delta-table.test.ts b/packages/cli/test/migration/1784000000043-create-workflow-statistics-delta-table.test.ts new file mode 100644 index 00000000000..ab5daad1a2f --- /dev/null +++ b/packages/cli/test/migration/1784000000043-create-workflow-statistics-delta-table.test.ts @@ -0,0 +1,125 @@ +import { + createTestMigrationContext, + initDbUpToMigration, + runSingleMigration, + undoLastSingleMigration, + type TestMigrationContext, +} from '@n8n/backend-test-utils'; +import { DbConnection } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { DataSource } from '@n8n/typeorm'; + +const MIGRATION_NAME = 'CreateWorkflowStatisticsDeltaTable1784000000043'; + +// The delta table + fold are Postgres-only; the migration is not registered for SQLite. +const runOnPostgres = (process.env.DB_TYPE ?? 'sqlite') === 'postgresdb'; +// eslint-disable-next-line n8n-local-rules/no-skipped-tests -- Postgres-only migration, skipped on SQLite +const describePg = runOnPostgres ? describe : describe.skip; + +describePg('CreateWorkflowStatisticsDeltaTable migration (Postgres)', () => { + let dataSource: DataSource; + + beforeAll(async () => { + const dbConnection = Container.get(DbConnection); + await dbConnection.init(); + dataSource = Container.get(DataSource); + }); + + beforeEach(async () => { + const context = createTestMigrationContext(dataSource); + await context.queryRunner.clearDatabase(); + await context.queryRunner.release(); + // Run every migration before this one (counter table exists; delta table does not). + await initDbUpToMigration(MIGRATION_NAME); + }); + + afterAll(async () => { + await Container.get(DbConnection).close(); + }); + + const insertCounter = async ( + context: TestMigrationContext, + row: { + workflowId: string; + name: string; + count: number; + rootCount: number; + workflowName: string; + }, + ) => { + const t = context.escape.tableName('workflow_statistics'); + const c = (n: string) => context.escape.columnName(n); + await context.runQuery( + `INSERT INTO ${t} (${c('workflowId')}, ${c('name')}, ${c('count')}, ${c('rootCount')}, ${c('latestEvent')}, ${c('workflowName')}) + VALUES (:workflowId, :name, :count, :rootCount, :latestEvent, :workflowName)`, + { ...row, latestEvent: new Date() }, + ); + }; + + const insertDelta = async ( + context: TestMigrationContext, + row: { workflowId: string; name: string; rootCountDelta: number; workflowName: string }, + ) => { + const t = context.escape.tableName('workflow_statistics_delta'); + const c = (n: string) => context.escape.columnName(n); + await context.runQuery( + `INSERT INTO ${t} (${c('workflowId')}, ${c('name')}, ${c('rootCountDelta')}, ${c('workflowName')}) + VALUES (:workflowId, :name, :rootCountDelta, :workflowName)`, + row, + ); + }; + + test('up creates the delta table; down folds residual deltas into the counter then drops it', async () => { + // up + await runSingleMigration(MIGRATION_NAME); + let context = createTestMigrationContext(dataSource); + + // A workflow already activated before rollback, plus three un-folded deltas (2 root, 1 non-root). + await insertCounter(context, { + workflowId: 'wf-1', + name: 'production_success', + count: 10, + rootCount: 4, + workflowName: 'WF One (old)', + }); + await insertDelta(context, { + workflowId: 'wf-1', + name: 'production_success', + rootCountDelta: 1, + workflowName: 'WF One', + }); + await insertDelta(context, { + workflowId: 'wf-1', + name: 'production_success', + rootCountDelta: 1, + workflowName: 'WF One', + }); + await insertDelta(context, { + workflowId: 'wf-1', + name: 'production_success', + rootCountDelta: 0, + workflowName: 'WF One', + }); + await context.queryRunner.release(); + + // down: must fold (count += 3, rootCount += 2) before dropping — no counts lost on rollback. + await undoLastSingleMigration(); + context = createTestMigrationContext(dataSource); + + const counterTable = context.escape.tableName('workflow_statistics'); + const countCol = context.escape.columnName('count'); + const rootCountCol = context.escape.columnName('rootCount'); + const wfIdCol = context.escape.columnName('workflowId'); + const counter = await context.runQuery>( + `SELECT ${countCol}::int AS count, ${rootCountCol}::int AS "rootCount" FROM ${counterTable} WHERE ${wfIdCol} = :id`, + { id: 'wf-1' }, + ); + expect(counter).toHaveLength(1); + expect(counter[0]).toMatchObject({ count: 13, rootCount: 6 }); + + // delta table is gone + const deltaTable = context.escape.tableName('workflow_statistics_delta'); + await expect(context.runQuery(`SELECT 1 FROM ${deltaTable}`)).rejects.toThrow(); + await context.queryRunner.release(); + }); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index dbb71aefe7d..86d6747f0a7 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -8606,6 +8606,7 @@ packages: '@langchain/community@1.1.27': resolution: {integrity: sha512-s2U3w7QV7QpkFtY1eZMni4poz+nKLFclpDi3a7hUbZ67ttsGaU9WkZ2BiLuzLIs+IFaUvON/KcGkE8EqAl9aPA==} engines: {node: '>=20'} + deprecated: This package has been deprecated. See https://github.com/langchain-ai/langchainjs-community/issues/61 for more info peerDependencies: '@arcjet/redact': ^v1.2.0 '@aws-crypto/sha256-js': ^5.0.0