mirror of
https://github.com/n8n-io/n8n.git
synced 2026-07-27 11:05:14 +02:00
perf(core): Aggregate workflow statistics to avoid hot-row contention (#33045)
This commit is contained in:
Iván Ovejero
GitHub
parent
b610b42558
commit
378710072f
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -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({});
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ export const LOG_SCOPES = [
|
|||
'task-runner-js',
|
||||
'task-runner-py',
|
||||
'insights',
|
||||
'workflow-statistics',
|
||||
'workflow-activation',
|
||||
'ssh-client',
|
||||
'data-table',
|
||||
|
|
|
|||
|
|
@ -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}`);
|
||||
}
|
||||
}
|
||||
|
|
@ -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,
|
||||
];
|
||||
|
|
|
|||
|
|
@ -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<WorkflowStatistics> {
|
||||
constructor(
|
||||
|
|
@ -65,58 +86,129 @@ export class WorkflowStatisticsRepository extends Repository<WorkflowStatistics>
|
|||
isRootExecution: boolean,
|
||||
workflowName?: string,
|
||||
): Promise<StatisticsUpsertResult> {
|
||||
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<void> {
|
||||
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<FoldRow[]>(
|
||||
`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<number> {
|
||||
const result = await this.createQueryBuilder('ws')
|
||||
.select('COUNT(DISTINCT ws.workflowId)', 'count')
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<z.infer<typeof flagsSchema>> {
|
|||
|
||||
Container.get(ExecutionsPruningService).init();
|
||||
Container.get(WorkflowHistoryCompactionService).init();
|
||||
Container.get(WorkflowStatisticsRollupService).init();
|
||||
Container.get(N8NCheckpointStorage).init();
|
||||
|
||||
if (this.globalConfig.executions.mode === 'regular') {
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<ReturnType<WorkflowStatisticsRepository['rollupIncrements']>>;
|
||||
|
||||
describe('WorkflowStatisticsRollupService', () => {
|
||||
const dbConnection = mock<DbConnection>({ connectionState: { migrated: true } });
|
||||
|
||||
const makeService = (opts: {
|
||||
isLeader?: boolean;
|
||||
instanceType?: 'main' | 'worker';
|
||||
dbType?: 'postgresdb' | 'sqlite';
|
||||
}) => {
|
||||
const instanceSettings = mock<InstanceSettings>({
|
||||
isLeader: opts.isLeader ?? true,
|
||||
instanceType: opts.instanceType ?? 'main',
|
||||
instanceRole: 'leader',
|
||||
});
|
||||
const databaseConfig = mock<DatabaseConfig>({ type: opts.dbType ?? 'postgresdb' });
|
||||
const errorReporter = mock<ErrorReporter>();
|
||||
const dbLockService = mock<DbLockService>();
|
||||
const repository = mock<WorkflowStatisticsRepository>();
|
||||
const statisticsService = mock<WorkflowStatisticsService>();
|
||||
const logger = mock<Logger>(); // the scoped logger the service actually logs to
|
||||
const service = new WorkflowStatisticsRollupService(
|
||||
mock<Logger>({ 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<number> }).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<RollupResult>((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);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -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<EntityManager>();
|
||||
const dataSource = mock<DataSource>({
|
||||
|
|
@ -566,6 +841,7 @@ describe('WorkflowStatisticsService', () => {
|
|||
eventService,
|
||||
settingsRepository,
|
||||
workflowRepository,
|
||||
databaseConfig,
|
||||
);
|
||||
globalConfig.diagnostics.enabled = true;
|
||||
globalConfig.deployment.type = 'n8n-testing';
|
||||
|
|
|
|||
174
packages/cli/src/services/workflow-statistics-rollup.service.ts
Normal file
174
packages/cli/src/services/workflow-statistics-rollup.service.ts
Normal file
|
|
@ -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<ReturnType<WorkflowStatisticsRepository['rollupIncrements']>>;
|
||||
|
||||
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<void> | 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<number> {
|
||||
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<RollupResult | null> {
|
||||
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<void> {
|
||||
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 });
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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<WorkflowStatisticsEv
|
|||
private readonly eventService: EventService,
|
||||
private readonly settingsRepository: SettingsRepository,
|
||||
private readonly workflowRepository: WorkflowRepository,
|
||||
private readonly databaseConfig: DatabaseConfig,
|
||||
) {
|
||||
super({ captureRejections: true });
|
||||
if ('SKIP_STATISTICS_EVENTS' in process.env) return;
|
||||
|
|
@ -115,29 +118,76 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
|
|||
const workflowId = workflowData.id;
|
||||
if (!workflowId) return;
|
||||
|
||||
const isRoot = isRootExecutionForRun(runData);
|
||||
|
||||
let upsertResult: Awaited<ReturnType<WorkflowStatisticsRepository['upsertWorkflowStatistics']>>;
|
||||
|
||||
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<void> {
|
||||
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<void> {
|
||||
const project = await this.ownershipService.getWorkflowProjectCached(workflowId);
|
||||
let userId: string | null = null;
|
||||
|
|
@ -150,7 +200,7 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
|
|||
await this.userService.updateSettings(owner.id, {
|
||||
firstSuccessfulWorkflowId: workflowId,
|
||||
userActivated: true,
|
||||
userActivatedAt: runData.startedAt.getTime(),
|
||||
userActivatedAt: userActivatedAtMs,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -163,9 +213,9 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
|
|||
}
|
||||
|
||||
private async emitInstanceFirstProductionWorkflowFailed(
|
||||
workflowData: IWorkflowBase,
|
||||
workflowId: string,
|
||||
runData: IRun,
|
||||
workflowName: string,
|
||||
timestampMs: number,
|
||||
): Promise<void> {
|
||||
const instanceHadProductionFailure = await this.settingsRepository.findByKey(
|
||||
'instance.firstProductionFailure',
|
||||
|
|
@ -193,7 +243,7 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
|
|||
workflowId,
|
||||
projectId: project.id,
|
||||
userId: owner.id,
|
||||
timestamp: runData.startedAt.getTime(),
|
||||
timestamp: timestampMs,
|
||||
}),
|
||||
loadOnStartup: false,
|
||||
});
|
||||
|
|
@ -201,7 +251,7 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
|
|||
this.eventService.emit('instance-first-production-workflow-failed', {
|
||||
projectId: project.id,
|
||||
workflowId,
|
||||
workflowName: workflowData.name,
|
||||
workflowName,
|
||||
userId: owner.id,
|
||||
});
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@
|
|||
*/
|
||||
|
||||
import { createTeamProject, createWorkflow, testDb, testModules } from '@n8n/backend-test-utils';
|
||||
import { GlobalConfig } from '@n8n/config';
|
||||
import type { Project, WorkflowEntity } from '@n8n/db';
|
||||
import { ExecutionRepository, StatisticsNames, WorkflowStatisticsRepository } from '@n8n/db';
|
||||
import { Container } from '@n8n/di';
|
||||
|
|
@ -147,8 +148,15 @@ describe('Insights vs Workflow Statistics Integration', () => {
|
|||
expectedCount: number,
|
||||
timeout = 10000,
|
||||
): Promise<void> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -62,10 +62,9 @@ describe('LicenseMetricsRepository', () => {
|
|||
StatisticsNames.manualError,
|
||||
secondWorkflow.id,
|
||||
),
|
||||
workflowStatisticsRepository.upsertWorkflowStatistics(
|
||||
workflowStatisticsRepository.insertWorkflowStatistics(
|
||||
StatisticsNames.productionSuccess,
|
||||
secondWorkflow.id,
|
||||
true,
|
||||
),
|
||||
]);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<Array<{ count: number; rootCount: number }>>(
|
||||
`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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user