github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 05:29:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

fix(core): Count executions that crash recovery marks as crashed (#37733)

Browse Source 
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Emilia 2026-09-08 11:29:15 +00:00 committed by GitHub
parent 6acdf815ea
commit 3521e54b81
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
20 changed files with 781 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

110
packages/@n8n/db/src/repositories/__tests__/execution.repository.test.ts
View File

@ -10,8 +10,9 @@ import type { IRunExecutionData, IWorkflowBase } from 'n8n-workflow';
import { nanoid } from 'nanoid';
import { mock } from 'vitest-mock-extended';
import { ExecutionEntity } from '../../entities';
import { ExecutionEntity, type WorkflowEntity } from '../../entities';
import type { IExecutionResponse } from '../../entities/types-db';
import { TransactionRunner } from '../../services/transaction';
import { mockEntityManager } from '../../utils/test-utils/mock-entity-manager';
import { mockInstance } from '../../utils/test-utils/mock-instance';
import { ExecutionRepository } from '../execution.repository';
@ -27,10 +28,13 @@ describe('ExecutionRepository', () => {
logging: { outputs: ['console'], scopes: [] },
});
mockInstance(BinaryDataService);
const transactionRunner = mock<TransactionRunner>();
Container.set(TransactionRunner, transactionRunner);
const executionRepository = Container.get(ExecutionRepository);
beforeEach(() => {
vi.resetAllMocks();
transactionRunner.run.mockImplementation(async (ctx, fn) => await fn(ctx));
});
describe('countInWorkflows', () => {
@ -245,19 +249,38 @@ describe('ExecutionRepository', () => {
});
});
describe('markAsCrashed', () => {
test('should batch updates above a threshold', async () => {
// Generates a list of many execution ids.
// NOTE: GREATER_THAN_MAX_UPDATE_THRESHOLD is selected to be just above the default threshold.
const manyExecutionsToMarkAsCrashed: string[] = Array(GREATER_THAN_MAX_UPDATE_THRESHOLD)
.fill(undefined)
.map((_, i) => i.toString());
await executionRepository.markAsCrashed(manyExecutionsToMarkAsCrashed);
expect(entityManager.update).toBeCalledTimes(2);
const crashableRow = (id: string) =>
mock<ExecutionEntity>({
id,
workflowId: `workflow-${id}`,
mode: 'trigger',
workflow: mock<WorkflowEntity>({ id: `workflow-${id}`, name: `Workflow ${id}` }),
});
test('should clear waitTill when marking executions as crashed', async () => {
const executionIdsOfLength = (length: number) =>
Array(length)
.fill(undefined)
.map((_, i) => i.toString());
describe('markAsCrashed', () => {
test('should batch updates above a threshold and report each batch as it transitions', async () => {
const manyExecutionsToMarkAsCrashed = executionIdsOfLength(GREATER_THAN_MAX_UPDATE_THRESHOLD);
entityManager.find.mockResolvedValue([crashableRow('1')]);
const reported: string[][] = [];
const crashed = await executionRepository.markAsCrashed(
manyExecutionsToMarkAsCrashed,
(batch) => reported.push(batch.map(({ id }) => id)),
);
expect(entityManager.update).toBeCalledTimes(2);
expect(reported).toEqual([['1'], ['1']]);
expect(crashed).toHaveLength(2);
});
test('should crash only in-progress rows and clear their `waitTill`', async () => {
const executionIds = ['1', '2'];
entityManager.find.mockResolvedValue(executionIds.map(crashableRow));
await executionRepository.markAsCrashed(executionIds);
@ -267,6 +290,71 @@ describe('ExecutionRepository', () => {
expect.objectContaining({ status: 'crashed', waitTill: null }),
);
});
test('should report the workflow, name and mode of each execution it transitioned', async () => {
entityManager.find.mockResolvedValue([crashableRow('1')]);
const crashed = await executionRepository.markAsCrashed(['1', '2']);
expect(crashed).toEqual([
{
id: '1',
workflowId: 'workflow-1',
workflowName: 'Workflow 1',
mode: 'trigger',
},
]);
});
test('should read back only the rows carrying the `stoppedAt` it wrote, soft-deleted ones included', async () => {
entityManager.find.mockResolvedValue([crashableRow('1')]);
await executionRepository.markAsCrashed(['1']);
const updateValues = entityManager.update.mock.calls[0][2] as { stoppedAt: Date };
const findOptions = entityManager.find.mock.calls[0][1];
expect(findOptions).toMatchObject({
where: { id: In(['1']), status: 'crashed', stoppedAt: updateValues.stoppedAt },
withDeleted: true,
});
});
test('should collapse duplicated ids before batching', async () => {
const oneBatchWorthOfIds = executionIdsOfLength(GREATER_THAN_MAX_UPDATE_THRESHOLD - 1);
entityManager.find.mockResolvedValue([crashableRow('1')]);
await executionRepository.markAsCrashed([...oneBatchWorthOfIds, '0']);
expect(entityManager.update).toBeCalledTimes(1);
});
test('should skip the read-back and report nothing when the UPDATE affects no rows', async () => {
entityManager.update.mockResolvedValue({ affected: 0, raw: [], generatedMaps: [] });
const crashed = await executionRepository.markAsCrashed(['1']);
expect(entityManager.find).not.toHaveBeenCalled();
expect(crashed).toEqual([]);
});
});
describe('markWorkflowExecutionsAsCrashed', () => {
test('should select the rows to crash by workflow rather than by id, and report each one it transitioned', async () => {
entityManager.find.mockResolvedValue([crashableRow('1')]);
const crashed = await executionRepository.markWorkflowExecutionsAsCrashed('workflow-1');
expect(entityManager.update).toBeCalledTimes(1);
expect(entityManager.update).toHaveBeenCalledWith(
ExecutionEntity,
{ workflowId: 'workflow-1', status: In(['new', 'running', 'unknown']) },
expect.objectContaining({ status: 'crashed', waitTill: null }),
);
expect(crashed).toEqual([
{ id: '1', workflowId: 'workflow-1', workflowName: 'Workflow 1', mode: 'trigger' },
]);
});
});
describe('setRunning', () => {

100
packages/@n8n/db/src/repositories/execution.repository.ts
View File

@ -22,6 +22,7 @@ import {
} from '@n8n/typeorm';
import { DateUtils } from '@n8n/typeorm/util/DateUtils';
import { stringify } from 'flatted';
import chunk from 'lodash/chunk';
import pick from 'lodash/pick';
import { BinaryDataService, ErrorReporter } from 'n8n-core';
import type {
@ -30,6 +31,7 @@ import type {
ExecutionSummary,
IRunExecutionData,
IRunExecutionDataAll,
WorkflowExecuteMode,
} from 'n8n-workflow';
import {
CRASHABLE_EXECUTION_STATUSES,
@ -67,6 +69,13 @@ class PostgresLiveRowsRetrievalError extends UnexpectedError {
}
}
export type CrashedExecution = {
id: string;
workflowId: string;
workflowName?: string;
mode: WorkflowExecuteMode;
};
export interface UpdateExecutionConditions {
requireStatus?: ExecutionStatus;
requireNotFinished?: boolean;
@ -373,23 +382,84 @@ export class ExecutionRepository extends BaseRepository<ExecutionEntity> {
} as IExecutionFlattedDb | IExecutionResponse | IExecutionBase;
}
async markAsCrashed(executionIds: string | string[]) {
if (!Array.isArray(executionIds)) executionIds = [executionIds];
/**
* Set in-progress executions to `crashed` in batches. Calls `onBatchTransitioned`
* after each batch commits.
*/
async markAsCrashed(
executionIds: string | string[],
onBatchTransitioned?: (batch: CrashedExecution[]) => void,
): Promise<CrashedExecution[]> {
// Dedupe so a repeated id is reported, and counted, once.
const ids = [...new Set(Array.isArray(executionIds) ? executionIds : [executionIds])];
let processed: number = 0;
while (processed < executionIds.length) {
// NOTE: if a slice goes past the end of the array, it just returns up til the end.
const batch: string[] = executionIds.slice(processed, processed + MAX_UPDATE_BATCH_SIZE);
await this.update(
// Guard against overwriting executions that have since moved to a `waiting` or
// terminal status: recovery can race a `running` -> `waiting` transition and flag a
// healthy execution as dangling, but only genuinely in-progress rows should be crashed
{ id: In(batch), status: In(CRASHABLE_EXECUTION_STATUSES) },
{ status: 'crashed', stoppedAt: new Date(), waitTill: null },
);
this.logger.info('Marked executions as `crashed`', { executionIds });
processed += batch.length;
const crashed: CrashedExecution[] = [];
for (const batch of chunk(ids, MAX_UPDATE_BATCH_SIZE)) {
const transitioned = await this.transitionToCrashed({ id: In(batch) });
crashed.push(...transitioned);
// Report each batch as it commits, so a later batch that throws keeps the earlier reports.
onBatchTransitioned?.(transitioned);
this.logger.info('Marked executions as `crashed`', { executionIds: batch });
}
return crashed;
}
/** Set the workflow's in-progress executions to `crashed`. */
async markWorkflowExecutionsAsCrashed(workflowId: string): Promise<CrashedExecution[]> {
const transitioned = await this.transitionToCrashed({ workflowId });
if (transitioned.length > 0) {
this.logger.info('Marked executions as `crashed`', {
executionIds: transitioned.map(({ id }) => id),
});
}
return transitioned;
}
/**
* Set the rows matching `where` to `crashed`. The caller's predicate selects the
* candidates; the status guard and the identity of the written rows are added here.
*/
private async transitionToCrashed(
where: FindOptionsWhere<ExecutionEntity>,
): Promise<CrashedExecution[]> {
// `stoppedAt` doubles as a claim token, so the read below can match the rows this
// UPDATE wrote without `SELECT ... FOR UPDATE`, which SQLite does not support. A
// batch that partly overlaps a concurrent sweep still reads the shared rows back.
const stoppedAt = new Date();
return await this.runInTransaction({}, async (tx) => {
// Guard against overwriting executions that have since moved to a `waiting` or
// terminal status: recovery can race a `running` -> `waiting` transition and flag a
// healthy execution as dangling, but only genuinely in-progress rows should be crashed
const updateResult = await tx.update(
ExecutionEntity,
{ ...where, status: In(CRASHABLE_EXECUTION_STATUSES) },
{ status: 'crashed', stoppedAt, waitTill: null },
);
// An unreported `affected` is unknown rather than zero, so it falls through to the read.
if (updateResult?.affected === 0) return [];
const rows = await tx.find(ExecutionEntity, {
select: { id: true, workflowId: true, mode: true, workflow: { id: true, name: true } },
relations: { workflow: true },
where: { ...where, status: 'crashed', stoppedAt },
// The UPDATE above also crashes soft-deleted rows, so keep them in the read.
withDeleted: true,
});
return rows.map(({ id, workflowId, mode, workflow }) => ({
id,
workflowId,
workflowName: workflow?.name,
mode,
}));
});
}
async setRunning(executionId: string) {

1
packages/@n8n/db/src/repositories/index.ts
View File

@ -35,6 +35,7 @@ export { ExecutionDataRepository } from './execution-data.repository';
export { ExecutionMetadataRepository } from './execution-metadata.repository';
export {
ExecutionRepository,
type CrashedExecution,
type ExecutionDeletionCriteria,
type UpdateExecutionConditions,
} from './execution.repository';

1
packages/cli/src/eventbus/message-event-bus/__tests__/message-event-bus.test.ts
View File

@ -50,6 +50,7 @@ describe('MessageEventBus.initialize', () => {
executionRepository,
mock(),
mock(),
mock(),
globalConfig,
mock<InstanceSettings>({ n8nFolder: tempDir }),
);

4
packages/cli/src/eventbus/message-event-bus/message-event-bus.ts
View File

@ -8,6 +8,7 @@ import uniqby from 'lodash/uniqBy';
import { InstanceSettings } from 'n8n-core';
import { existsSync } from 'node:fs';
import { ExecutionCrashService } from '../../executions/execution-crash.service';
import { ExecutionRecoveryService } from '../../executions/execution-recovery.service';
import type { EventMessageTypes } from '../event-message-classes/';
import {
@ -62,6 +63,7 @@ export class MessageEventBus extends EventEmitter {
private readonly executionRepository: ExecutionRepository,
private readonly workflowRepository: WorkflowRepository,
private readonly recoveryService: ExecutionRecoveryService,
private readonly executionCrashService: ExecutionCrashService,
private readonly globalConfig: GlobalConfig,
private readonly instanceSettings: InstanceSettings,
) {
@ -307,7 +309,7 @@ export class MessageEventBus extends EventEmitter {
const recoveryAlreadyAttempted = this.logWriter?.isRecoveryProcessRunning();
if (recoveryAlreadyAttempted || this.globalConfig.eventBus.crashRecoveryMode === 'simple') {
await this.executionRepository.markAsCrashed(unfinishedExecutionIds);
await this.executionCrashService.markAsCrashed(unfinishedExecutionIds);
// if we end up here, it means that the previous recovery process did not finish
// a possible reason would be that recreating the workflow data itself caused e.g an OOM error
// in that case, we do not want to retry the recovery process, but rather mark the executions as crashed

19
packages/cli/src/executions/__tests__/enqueued-execution-recovery.service.test.ts
View File

@ -1,12 +1,13 @@
import { mockLogger } from '@n8n/backend-test-utils';
import type { ExecutionsConfig } from '@n8n/config';
import type { ExecutionRepository, IExecutionResponse, Project } from '@n8n/db';
import type { IExecutionResponse, Project } from '@n8n/db';
import type { ErrorReporter } from 'n8n-core';
import { mock } from 'vitest-mock-extended';
import { ExecutionAlreadyResumingError } from '@/errors/execution-already-resuming.error';
import type { EventService } from '@/events/event.service';
import { EnqueuedExecutionRecoveryService } from '@/executions/enqueued-execution-recovery.service';
import type { ExecutionCrashService } from '@/executions/execution-crash.service';
import type { ExecutionService } from '@/executions/execution.service';
import type { OwnershipService } from '@/services/ownership.service';
import type { WorkflowRunner } from '@/workflow-runner';
@ -21,7 +22,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
vi.mocked(logger.scoped).mockReturnValue(logger);
const errorReporter = mock<ErrorReporter>();
const executionService = mock<ExecutionService>();
const executionRepository = mock<ExecutionRepository>();
const executionCrashService = mock<ExecutionCrashService>();
const ownershipService = mock<OwnershipService>();
const workflowRunner = mock<WorkflowRunner>();
const eventService = mock<EventService>();
@ -32,7 +33,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
errorReporter,
mock<ExecutionsConfig>({ mode }),
executionService,
executionRepository,
executionCrashService,
ownershipService,
workflowRunner,
eventService,
@ -83,7 +84,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
await createService().recoverEnqueuedExecutions();
await new Promise(setImmediate); // `run` is not awaited, let the rejection settle
expect(executionRepository.markAsCrashed).toHaveBeenCalledExactlyOnceWith('1');
expect(executionCrashService.markAsCrashed).toHaveBeenCalledExactlyOnceWith('1');
expect(errorReporter.error).toHaveBeenCalledTimes(1);
expect(workflowRunner.run).toHaveBeenCalledTimes(2);
});
@ -95,7 +96,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
await createService().recoverEnqueuedExecutions();
await new Promise(setImmediate); // `run` is not awaited, let the rejection settle
expect(executionRepository.markAsCrashed).not.toHaveBeenCalled();
expect(executionCrashService.markAsCrashed).not.toHaveBeenCalled();
expect(errorReporter.error).not.toHaveBeenCalled();
});
@ -106,7 +107,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
await createService().recoverEnqueuedExecutions();
expect(executionRepository.markAsCrashed).toHaveBeenCalledExactlyOnceWith(['2', '3']);
expect(executionCrashService.markAsCrashed).toHaveBeenCalledExactlyOnceWith(['2', '3']);
expect(workflowRunner.run).toHaveBeenCalledExactlyOnceWith(
expect.anything(),
undefined,
@ -120,7 +121,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
await createService().recoverEnqueuedExecutions();
expect(executionRepository.markAsCrashed).toHaveBeenCalledExactlyOnceWith(['1']);
expect(executionCrashService.markAsCrashed).toHaveBeenCalledExactlyOnceWith(['1']);
expect(workflowRunner.run).not.toHaveBeenCalled();
});
@ -129,7 +130,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
await createService().recoverEnqueuedExecutions();
expect(executionRepository.markAsCrashed).not.toHaveBeenCalled();
expect(executionCrashService.markAsCrashed).not.toHaveBeenCalled();
});
// A throw used to abort the whole loop, leaving every remaining execution at `new`.
@ -141,7 +142,7 @@ describe('EnqueuedExecutionRecoveryService', () => {
await createService().recoverEnqueuedExecutions();
expect(executionRepository.markAsCrashed).toHaveBeenCalledExactlyOnceWith('1');
expect(executionCrashService.markAsCrashed).toHaveBeenCalledExactlyOnceWith('1');
expect(workflowRunner.run).toHaveBeenCalledExactlyOnceWith(
expect.anything(),
undefined,

100
packages/cli/src/executions/__tests__/execution-crash.service.test.ts Normal file
View File

@ -0,0 +1,100 @@
import type { CrashedExecution, ExecutionRepository } from '@n8n/db';
import { mock } from 'vitest-mock-extended';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import type { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
describe('ExecutionCrashService', () => {
const executionRepository = mock<ExecutionRepository>();
const workflowStatisticsService = mock<WorkflowStatisticsService>();
const crashService = new ExecutionCrashService(executionRepository, workflowStatisticsService);
const crashedExecution = (id: string, overrides: Partial<CrashedExecution> = {}) => ({
id,
workflowId: `workflow-${id}`,
workflowName: `Workflow ${id}`,
mode: 'trigger' as const,
...overrides,
});
const transitions = (batch: CrashedExecution[]) =>
executionRepository.markAsCrashed.mockImplementation(async (_ids, onBatchTransitioned) => {
onBatchTransitioned?.(batch);
return await Promise.resolve(batch);
});
const emitted = () => workflowStatisticsService.emit.mock.calls;
beforeEach(() => {
vi.clearAllMocks();
});
test('reports the executions it transitioned for counting', async () => {
const first = crashedExecution('1');
const second = crashedExecution('2', { mode: 'manual' });
transitions([first, second]);
const crashed = await crashService.markAsCrashed(['1', '2']);
expect(crashed).toEqual([first, second]);
expect(emitted()).toEqual([['executionsCrashed', { executions: [first, second] }]]);
});
test('reports each batch as it transitions, so a later failure keeps earlier ones counted', async () => {
const first = crashedExecution('1');
const second = crashedExecution('2');
executionRepository.markAsCrashed.mockImplementation(async (_ids, onBatchTransitioned) => {
onBatchTransitioned?.([first]);
onBatchTransitioned?.([second]);
return await Promise.reject(new Error('connection reset'));
});
await expect(crashService.markAsCrashed(['1', '2', '3'])).rejects.toThrow('connection reset');
expect(emitted()).toEqual([
['executionsCrashed', { executions: [first] }],
['executionsCrashed', { executions: [second] }],
]);
});
test('reports nothing when no execution transitioned', async () => {
transitions([]);
const crashed = await crashService.markAsCrashed(['1']);
expect(crashed).toEqual([]);
expect(workflowStatisticsService.emit).not.toHaveBeenCalled();
});
test('counts nothing when the caller counts the executions itself', async () => {
transitions([crashedExecution('1')]);
await crashService.markAsCrashedWithoutCounting('1');
expect(executionRepository.markAsCrashed).toHaveBeenCalledWith('1');
expect(workflowStatisticsService.emit).not.toHaveBeenCalled();
});
test('reports the executions it transitioned for a whole workflow', async () => {
const first = crashedExecution('1');
const second = crashedExecution('2');
executionRepository.markWorkflowExecutionsAsCrashed.mockResolvedValue([first, second]);
const crashed = await crashService.markWorkflowExecutionsAsCrashed('workflow-1');
expect(crashed).toEqual([first, second]);
expect(emitted()).toEqual([['executionsCrashed', { executions: [first, second] }]]);
});
test('reports nothing when a workflow has no execution to transition', async () => {
executionRepository.markWorkflowExecutionsAsCrashed.mockResolvedValue([]);
const crashed = await crashService.markWorkflowExecutionsAsCrashed('workflow-1');
expect(crashed).toEqual([]);
expect(workflowStatisticsService.emit).not.toHaveBeenCalled();
});
});

2
packages/cli/src/executions/__tests__/execution-recovery.service.integration.test.ts
View File

@ -28,6 +28,7 @@ import { NodeCrashedError } from '@/errors/node-crashed.error';
import { WorkflowCrashedError } from '@/errors/workflow-crashed.error';
import type { EventMessageTypes as EventMessage } from '@/eventbus/event-message-classes';
import { EventMessageNode } from '@/eventbus/event-message-classes/event-message-node';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import { ExecutionPersistence } from '@/executions/execution-persistence';
import { ExecutionRecoveryService } from '@/executions/execution-recovery.service';
import { ExternalHooks } from '@/external-hooks';
@ -77,6 +78,7 @@ describe('ExecutionRecoveryService', () => {
ownershipService,
projectRelationRepository,
workflowPushNotifier,
new ExecutionCrashService(executionRepository, mock()),
);
});

8
packages/cli/src/executions/enqueued-execution-recovery.service.ts
View File

@ -1,6 +1,5 @@
import { Logger } from '@n8n/backend-common';
import { ExecutionsConfig } from '@n8n/config';
import { ExecutionRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import { ErrorReporter } from 'n8n-core';
import type { IWorkflowExecutionDataProcess } from 'n8n-workflow';
@ -8,6 +7,7 @@ import { strict as assert } from 'node:assert';
import { ExecutionAlreadyResumingError } from '@/errors/execution-already-resuming.error';
import { EventService } from '@/events/event.service';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import { ExecutionService } from '@/executions/execution.service';
import { OwnershipService } from '@/services/ownership.service';
import { WorkflowRunner } from '@/workflow-runner';
@ -23,7 +23,7 @@ export class EnqueuedExecutionRecoveryService {
private readonly errorReporter: ErrorReporter,
private readonly executionsConfig: ExecutionsConfig,
private readonly executionService: ExecutionService,
private readonly executionRepository: ExecutionRepository,
private readonly executionCrashService: ExecutionCrashService,
private readonly ownershipService: OwnershipService,
private readonly workflowRunner: WorkflowRunner,
private readonly eventService: EventService,
@ -54,7 +54,7 @@ export class EnqueuedExecutionRecoveryService {
this.logger.warn('Crashing enqueued executions with unreadable data', {
executionIds: unreadableIds,
});
await this.executionRepository.markAsCrashed(unreadableIds);
await this.executionCrashService.markAsCrashed(unreadableIds);
}
if (executions.length === 0) return;
@ -101,6 +101,6 @@ export class EnqueuedExecutionRecoveryService {
this.errorReporter.error(error, { executionId, shouldBeLogged: false });
this.logger.error('Failed to run enqueued execution', { executionId, error });
await this.executionRepository.markAsCrashed(executionId);
await this.executionCrashService.markAsCrashed(executionId);
}
}

43
packages/cli/src/executions/execution-crash.service.ts Normal file
View File

@ -0,0 +1,43 @@
import type { CrashedExecution } from '@n8n/db';
import { ExecutionRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
/**
* Marks executions as `crashed`. A crash transition runs no execution lifecycle
* hooks, so the workflow statistics are counted from here instead.
*/
@Service()
export class ExecutionCrashService {
constructor(
private readonly executionRepository: ExecutionRepository,
private readonly workflowStatisticsService: WorkflowStatisticsService,
) {}
async markAsCrashed(executionIds: string | string[]): Promise<CrashedExecution[]> {
return await this.executionRepository.markAsCrashed(executionIds, (batch) => this.count(batch));
}
/**
* Mark executions as `crashed` without counting them. For a caller that fires
* `workflowExecuteAfter` for the same executions, which counts them itself.
*/
async markAsCrashedWithoutCounting(executionIds: string | string[]): Promise<void> {
await this.executionRepository.markAsCrashed(executionIds);
}
async markWorkflowExecutionsAsCrashed(workflowId: string): Promise<CrashedExecution[]> {
const crashed = await this.executionRepository.markWorkflowExecutionsAsCrashed(workflowId);
this.count(crashed);
return crashed;
}
private count(executions: CrashedExecution[]) {
if (executions.length === 0) return;
this.workflowStatisticsService.emit('executionsCrashed', { executions });
}
}

19
packages/cli/src/executions/execution-recovery.service.ts
View File

@ -1,12 +1,6 @@
import { Logger } from '@n8n/backend-common';
import { ExecutionsConfig } from '@n8n/config';
import {
In,
type IExecutionResponse,
ProjectRelationRepository,
WorkflowEntity,
User,
} from '@n8n/db';
import { type IExecutionResponse, ProjectRelationRepository, WorkflowEntity, User } from '@n8n/db';
import { ExecutionRepository, WorkflowRepository } from '@n8n/db';
import { Container, Service } from '@n8n/di';
import { PROJECT_ADMIN_ROLE_SLUG, PROJECT_OWNER_ROLE_SLUG } from '@n8n/permissions';
@ -15,12 +9,13 @@ import { sleep } from '@n8n/utils/sleep';
import type { DateTime } from 'luxon';
import { InstanceSettings } from 'n8n-core';
import { createEmptyRunExecutionData } from 'n8n-workflow';
import { ExecutionStatus, type IRun, type ITaskData } from 'n8n-workflow';
import type { IRun, ITaskData } from 'n8n-workflow';
import { ARTIFICIAL_TASK_DATA } from '@/constants';
import { NodeCrashedError } from '@/errors/node-crashed.error';
import { WorkflowCrashedError } from '@/errors/workflow-crashed.error';
import { getLifecycleHooksForRegularMain } from '@/execution-lifecycle/execution-lifecycle-hooks';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import { ExecutionPersistence } from '@/executions/execution-persistence';
import { Push } from '@/push';
import { OwnershipService } from '@/services/ownership.service';
@ -46,6 +41,7 @@ export class ExecutionRecoveryService {
private readonly ownershipService: OwnershipService,
private readonly projectRelationRepository: ProjectRelationRepository,
private readonly workflowPushNotifier: WorkflowPushNotifier,
private readonly executionCrashService: ExecutionCrashService,
) {}
async autoDeactivateWorkflowsIfNeeded(workflowIds: Set<string>) {
@ -106,10 +102,7 @@ export class ExecutionRecoveryService {
}
}
await this.executionRepository.update(
{ workflowId, status: In<ExecutionStatus>(['running', 'new']) },
{ status: 'crashed', stoppedAt: new Date() },
);
await this.executionCrashService.markWorkflowExecutionsAsCrashed(workflowId);
}
}
}
@ -237,7 +230,7 @@ export class ExecutionRecoveryService {
if (!exists) return null;
await this.executionRepository.markAsCrashed(executionId);
await this.executionCrashService.markAsCrashedWithoutCounting(executionId);
const execution = await this.executionPersistence.findSingleExecution(executionId, {
includeData: true,

13
packages/cli/src/scaling/__tests__/scaling.service.test.ts
View File

@ -11,6 +11,7 @@ import type { MockInstance } from 'vitest';
import { mock } from 'vitest-mock-extended';
import type { ActiveExecutions } from '@/active-executions';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import type { ExecutionPersistence } from '@/executions/execution-persistence';
import { JOB_TYPE_NAME } from '../constants';
@ -100,6 +101,7 @@ describe('ScalingService', () => {
const jobProcessor = mock<JobProcessor>();
const executionRepository = mock<ExecutionRepository>();
const executionPersistence = mock<ExecutionPersistence>();
const executionCrashService = mockInstance(ExecutionCrashService);
const webhookResponseRelay = mock<WebhookResponseRelay>();
let scalingService: ScalingService;
@ -143,6 +145,7 @@ describe('ScalingService', () => {
instanceSettings,
mock(),
webhookResponseRelay,
executionCrashService,
);
getRunningJobsCountSpy = vi.spyOn(scalingService, 'getRunningJobsCount');
@ -743,6 +746,7 @@ describe('ScalingService', () => {
instanceSettings,
mock(),
webhookResponseRelay,
executionCrashService,
);
await scalingService.setupQueue();
@ -781,6 +785,7 @@ describe('ScalingService', () => {
instanceSettings,
mock(),
webhookResponseRelay,
executionCrashService,
);
await scalingService.setupQueue();
@ -814,6 +819,7 @@ describe('ScalingService', () => {
instanceSettings,
mock(),
webhookResponseRelay,
executionCrashService,
);
await scalingService.setupQueue();
@ -850,6 +856,7 @@ describe('ScalingService', () => {
instanceSettings,
mock(),
webhookResponseRelay,
executionCrashService,
);
await scalingService.setupQueue();
@ -891,7 +898,7 @@ describe('ScalingService', () => {
await scalingService.recoverFromQueue();
expect(executionRepository.markAsCrashed).toHaveBeenCalledWith(['123']);
expect(executionCrashService.markAsCrashed).toHaveBeenCalledWith(['123']);
});
it('should mark running executions as crashed if they are missing from the queue and queue is not empty', async () => {
@ -901,7 +908,7 @@ describe('ScalingService', () => {
await scalingService.recoverFromQueue();
expect(executionRepository.markAsCrashed).toHaveBeenCalledWith(['123']);
expect(executionCrashService.markAsCrashed).toHaveBeenCalledWith(['123']);
});
it('should not mark running executions as crashed if they are present in the queue', async () => {
@ -911,7 +918,7 @@ describe('ScalingService', () => {
await scalingService.recoverFromQueue();
expect(executionRepository.markAsCrashed).not.toHaveBeenCalled();
expect(executionCrashService.markAsCrashed).not.toHaveBeenCalled();
});
});

4
packages/cli/src/scaling/scaling.service.ts
View File

@ -14,6 +14,7 @@ import assert, { strict } from 'node:assert';
import { ActiveExecutions } from '@/active-executions';
import { HIGHEST_SHUTDOWN_PRIORITY } from '@/constants';
import { EventService } from '@/events/event.service';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import { ExecutionPersistence } from '@/executions/execution-persistence';
import { assertNever } from '@/utils';
@ -64,6 +65,7 @@ export class ScalingService {
private readonly instanceSettings: InstanceSettings,
private readonly eventService: EventService,
private readonly webhookResponseRelay: WebhookResponseRelay,
private readonly executionCrashService: ExecutionCrashService,
) {
this.logger = this.logger.scoped('scaling');
}
@ -778,7 +780,7 @@ export class ScalingService {
return waitMs;
}
await this.executionRepository.markAsCrashed(danglingIds);
await this.executionCrashService.markAsCrashed(danglingIds);
this.logger.info('Completed queue recovery check, recovered dangling executions', {
danglingIds,

81
packages/cli/src/services/workflow-statistics.service.ts
View File

@ -1,5 +1,6 @@
import { Logger, TypedEmitter } from '@n8n/backend-common';
import { DatabaseConfig } from '@n8n/config';
import type { CrashedExecution } from '@n8n/db';
import {
SettingsRepository,
StatisticsNames,
@ -60,8 +61,13 @@ const isModeRootExecution = {
agent: false,
} satisfies Record<WorkflowExecuteMode, boolean>;
type CompletedRunOutcome = {
mode: WorkflowExecuteMode;
status: ExecutionStatus;
};
function getStatisticsNameForCompletedRun(
runData: IRun,
runData: CompletedRunOutcome,
source?: WorkflowExecutionSource,
): StatisticsNames | null {
const isChatExecution = runData.mode === 'chat';
@ -84,7 +90,7 @@ function getStatisticsNameForCompletedRun(
: StatisticsNames.productionError;
}
function isRootExecutionForRun(runData: IRun): boolean {
function isRootExecutionForRun(runData: CompletedRunOutcome): boolean {
return isModeRootExecution[runData.mode] && isStatusRootExecution[runData.status];
}
@ -95,6 +101,7 @@ type WorkflowStatisticsEvents = {
fullRunData: IRun;
source?: WorkflowExecutionSource;
};
executionsCrashed: { executions: CrashedExecution[] };
};
@Service()
@ -121,6 +128,12 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
async ({ workflowData, fullRunData, source }) =>
await this.workflowExecutionCompleted(workflowData, fullRunData, source),
);
this.on('executionsCrashed', async ({ executions }) => {
// Record one at a time, to keep a large sweep within the database connection pool.
for (const { workflowId, workflowName, mode } of executions) {
await this.recordExecutionOutcome({ workflowId, workflowName, mode, status: 'crashed' });
}
});
}
async workflowExecutionCompleted(
@ -138,6 +151,57 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
const workflowId = workflowData.id;
if (!workflowId) return;
await this.recordCompletedRun({
statisticsName,
workflowId,
workflowName: workflowData.name,
isRoot,
firstEventMs: runData.startedAt.getTime(),
});
}
private async recordExecutionOutcome({
workflowId,
workflowName,
mode,
status,
}: { workflowId: string; workflowName?: string } & CompletedRunOutcome): Promise<void> {
// Contain every failure: this runs from an emitter listener that captures
// rejections and has no `error` handler, so a rejection would end the process.
try {
const outcome = { mode, status };
const statisticsName = getStatisticsNameForCompletedRun(outcome);
if (!statisticsName) return;
await this.recordCompletedRun({
statisticsName,
workflowId,
workflowName,
isRoot: isRootExecutionForRun(outcome),
firstEventMs: Date.now(),
});
} catch (error) {
this.logger.error('Failed to record the outcome of an execution', {
workflowId,
error: ensureError(error),
});
}
}
private async recordCompletedRun({
statisticsName,
workflowId,
workflowName,
isRoot,
firstEventMs,
}: {
statisticsName: StatisticsNames;
workflowId: string;
workflowName?: string;
isRoot: boolean;
firstEventMs: number;
}): Promise<void> {
let upsertResult: Awaited<ReturnType<WorkflowStatisticsRepository['upsertWorkflowStatistics']>>;
try {
@ -146,12 +210,7 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
* whereas in SQLite we upsert directly.
*/
if (this.databaseConfig.type === 'postgresdb') {
await this.repository.appendIncrement(
statisticsName,
workflowId,
isRoot,
workflowData.name,
);
await this.repository.appendIncrement(statisticsName, workflowId, isRoot, workflowName);
return;
}
@ -159,7 +218,7 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
statisticsName,
workflowId,
isRoot,
workflowData.name,
workflowName,
);
} catch (error) {
this.logger.error('Failed to record workflow statistic', { error: ensureError(error) });
@ -172,8 +231,8 @@ export class WorkflowStatisticsService extends TypedEmitter<WorkflowStatisticsEv
await this.emitFirstOccurrenceEvent(
statisticsName,
workflowId,
workflowData.name ?? null,
runData.startedAt.getTime(),
workflowName ?? null,
firstEventMs,
);
} catch (error) {
this.logger.debug('Failed to emit workflow statistics milestone', {

10
packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts
View File

@ -2,7 +2,6 @@ import type { Logger } from '@n8n/backend-common';
import type { GlobalConfig, WorkflowsConfig } from '@n8n/config';
import type {
CreateExecutionPayload,
ExecutionRepository,
Project,
User,
WorkflowEntity,
@ -33,6 +32,7 @@ import { DuplicateExecutionError } from '@/errors/duplicate-execution.error';
import { ExecutionAlreadyResumingError } from '@/errors/execution-already-resuming.error';
import { PreExecuteBlockedError } from '@/errors/pre-execute-blocked.error';
import type { EventService } from '@/events/event.service';
import type { ExecutionCrashService } from '@/executions/execution-crash.service';
import type { IWorkflowErrorData } from '@/interfaces';
import type { NodeTypes } from '@/node-types';
import type { OwnershipService } from '@/services/ownership.service';
@ -107,7 +107,7 @@ describe('WorkflowExecutionService', () => {
const nodeTypes = mock<NodeTypes>();
const workflowRunner = mock<WorkflowRunner>();
const pollCursorService = mock<PollCursorService>();
const executionRepository = mock<ExecutionRepository>();
const executionCrashService = mock<ExecutionCrashService>();
const logger = mock<Logger>();
const errorReporter = mock<ErrorReporter>();
const workflowExecutionService = new WorkflowExecutionService(
@ -127,7 +127,7 @@ describe('WorkflowExecutionService', () => {
mock(),
mock(),
pollCursorService,
executionRepository,
executionCrashService,
);
const additionalData = mock<IWorkflowExecuteAdditionalData>({});
@ -364,7 +364,7 @@ describe('WorkflowExecutionService', () => {
const returned = await runPolledWorkflow();
expect(returned).toBe('exec-9');
expect(executionRepository.markAsCrashed).toHaveBeenCalledWith('exec-9');
expect(executionCrashService.markAsCrashed).toHaveBeenCalledWith('exec-9');
expect(responsePromise.reject).toHaveBeenCalledWith(runError);
expect(errorReporter.error).toHaveBeenCalledWith(runError, expect.anything());
});
@ -375,7 +375,7 @@ describe('WorkflowExecutionService', () => {
const returned = await runPolledWorkflow();
expect(returned).toBe('exec-9');
expect(executionRepository.markAsCrashed).not.toHaveBeenCalled();
expect(executionCrashService.markAsCrashed).not.toHaveBeenCalled();
expect(responsePromise.reject).not.toHaveBeenCalled();
});

7
packages/cli/src/workflows/workflow-execution.service.ts
View File

@ -7,7 +7,7 @@ import type {
WorkflowEntity,
PollLeaseFence,
} from '@n8n/db';
import { ExecutionRepository, WorkflowRepository } from '@n8n/db';
import { WorkflowRepository } from '@n8n/db';
import { Service } from '@n8n/di';
import { ensureError } from '@n8n/utils/errors/ensure-error';
import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise';
@ -42,6 +42,7 @@ import {
import { ExecutionAlreadyResumingError } from '@/errors/execution-already-resuming.error';
import { PreExecuteBlockedError } from '@/errors/pre-execute-blocked.error';
import { EventService } from '@/events/event.service';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import { ExecutionPersistence } from '@/executions/execution-persistence';
import { FailedRunFactory } from '@/executions/failed-run-factory';
import { SubworkflowPolicyChecker } from '@/executions/pre-execution-checks';
@ -75,7 +76,7 @@ export class WorkflowExecutionService {
private readonly workflowsConfig: WorkflowsConfig,
private readonly workflowPublishedDataService: WorkflowPublishedDataService,
private readonly pollCursorService: PollCursorService,
private readonly executionRepository: ExecutionRepository,
private readonly executionCrashService: ExecutionCrashService,
) {}
async runWorkflow(
@ -275,7 +276,7 @@ export class WorkflowExecutionService {
responsePromise?.reject(ensureError(error));
await this.executionRepository.markAsCrashed(executionId);
await this.executionCrashService.markAsCrashed(executionId);
}
private isDestinationNodeATrigger(destinationNode: string, workflow: IWorkflowBase) {

47
packages/cli/test/integration/database/repositories/execution.repository.test.ts
View File

@ -197,14 +197,15 @@ describe('ExecutionRepository', () => {
createdAt: new Date(),
...extra,
});
return identifiers[0].id as string;
// Postgres returns the inserted id as a number, SQLite as a string.
return { id: String(identifiers[0].id), workflow };
};
it('should crash in-progress and indeterminate executions', async () => {
const executionRepo = Container.get(ExecutionRepository);
const newId = await createExecution('new');
const runningId = await createExecution('running');
const unknownId = await createExecution('unknown');
const { id: newId } = await createExecution('new');
const { id: runningId } = await createExecution('running');
const { id: unknownId } = await createExecution('unknown');
await executionRepo.markAsCrashed([newId, runningId, unknownId]);
@ -222,7 +223,7 @@ describe('ExecutionRepository', () => {
it('should not overwrite a waiting execution or clear its waitTill', async () => {
const executionRepo = Container.get(ExecutionRepository);
const waitTill = new Date(Date.now() + 1000 * 60 * 60);
const waitingId = await createExecution('waiting', { waitTill });
const { id: waitingId } = await createExecution('waiting', { waitTill });
await executionRepo.markAsCrashed([waitingId]);
@ -233,10 +234,10 @@ describe('ExecutionRepository', () => {
it('should not overwrite executions in a terminal status', async () => {
const executionRepo = Container.get(ExecutionRepository);
const successId = await createExecution('success');
const errorId = await createExecution('error');
const canceledId = await createExecution('canceled');
const crashedId = await createExecution('crashed');
const { id: successId } = await createExecution('success');
const { id: errorId } = await createExecution('error');
const { id: canceledId } = await createExecution('canceled');
const { id: crashedId } = await createExecution('crashed');
await executionRepo.markAsCrashed([successId, errorId, canceledId, crashedId]);
@ -256,9 +257,9 @@ describe('ExecutionRepository', () => {
it('should crash only the crashable executions in a mixed batch', async () => {
const executionRepo = Container.get(ExecutionRepository);
const waitTill = new Date(Date.now() + 1000 * 60 * 60);
const runningId = await createExecution('running');
const waitingId = await createExecution('waiting', { waitTill });
const successId = await createExecution('success');
const { id: runningId } = await createExecution('running');
const { id: waitingId } = await createExecution('waiting', { waitTill });
const { id: successId } = await createExecution('success');
await executionRepo.markAsCrashed([runningId, waitingId, successId]);
@ -278,6 +279,28 @@ describe('ExecutionRepository', () => {
expect(waitingExec?.waitTill?.getTime()).toBe(waitTill.getTime());
expect(successExec?.status).toBe('success');
});
it('should crash a soft-deleted in-progress execution', async () => {
const executionRepo = Container.get(ExecutionRepository);
const { id: runningId, workflow } = await createExecution('running');
await executionRepo.softDelete(runningId);
const crashed = await executionRepo.markAsCrashed([runningId]);
const runningExec = await executionRepo.findOne({
where: { id: runningId },
withDeleted: true,
});
expect(runningExec?.status).toBe('crashed');
expect(crashed).toEqual([
{
id: runningId,
workflowId: workflow.id,
workflowName: workflow.name,
mode: 'manual',
},
]);
});
});
describe('cancelManyRunning', () => {

86
packages/cli/test/integration/executions/execution-recovery-statistics.test.ts Normal file
View File

@ -0,0 +1,86 @@
import { createWorkflow, mockLogger, testDb } from '@n8n/backend-test-utils';
import { GlobalConfig } from '@n8n/config';
import type { WorkflowEntity } from '@n8n/db';
import { ExecutionRepository, StatisticsNames, WorkflowRepository } from '@n8n/db';
import { Container } from '@n8n/di';
import { mock } from 'vitest-mock-extended';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import { ExecutionRecoveryService } from '@/executions/execution-recovery.service';
import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
import { createExecution } from '../shared/db/executions';
import { findWorkflowStatistic, useWorkflowStatisticsEvents } from '../shared/workflow-statistics';
describe('ExecutionRecoveryService auto-deactivation', () => {
let recoveryService: ExecutionRecoveryService;
let maxLastExecutions: number;
let workflow: WorkflowEntity;
useWorkflowStatisticsEvents();
beforeAll(async () => {
await testDb.init();
Container.get(WorkflowStatisticsService);
const globalConfig = Container.get(GlobalConfig);
maxLastExecutions = globalConfig.executions.recovery.maxLastExecutions;
recoveryService = new ExecutionRecoveryService(
mockLogger(),
mock(),
mock(),
Container.get(ExecutionRepository),
mock(),
globalConfig.executions,
Container.get(WorkflowRepository),
mock(),
mock(),
mock(),
mock(),
Container.get(ExecutionCrashService),
);
});
beforeEach(async () => {
await testDb.truncate([
'WorkflowEntity',
'WorkflowStatistics',
'WorkflowStatisticsDelta',
'ExecutionEntity',
]);
workflow = await createWorkflow();
});
afterAll(async () => {
await testDb.terminate();
});
it('records a production error for an execution left in progress by a deactivated workflow', async () => {
for (let i = 0; i < maxLastExecutions; i++) {
await createExecution({ status: 'crashed' }, workflow);
}
// older than the crashed ones, so it stays outside the last-N window
await createExecution(
{
status: 'running',
finished: false,
stoppedAt: null,
mode: 'trigger',
startedAt: new Date(Date.now() - 60_000),
},
workflow,
);
await recoveryService.autoDeactivateWorkflowsIfNeeded(new Set([workflow.id]));
await vi.waitFor(async () => {
expect(
await findWorkflowStatistic(workflow.id, StatisticsNames.productionError),
).toMatchObject({ count: 1, rootCount: 1, workflowName: workflow.name });
});
});
});

171
packages/cli/test/integration/scaling/queue-recovery.test.ts Normal file
View File

@ -0,0 +1,171 @@
import { createWorkflow, mockLogger, testDb } from '@n8n/backend-test-utils';
import { GlobalConfig } from '@n8n/config';
import type { WorkflowEntity } from '@n8n/db';
import { ExecutionRepository, StatisticsNames } from '@n8n/db';
import { Container } from '@n8n/di';
import { InstanceSettings } from 'n8n-core';
import type { WorkflowExecuteMode } from 'n8n-workflow';
import { mock } from 'vitest-mock-extended';
import { ExecutionCrashService } from '@/executions/execution-crash.service';
import { ScalingService } from '@/scaling/scaling.service';
import { WorkflowStatisticsService } from '@/services/workflow-statistics.service';
import { createExecution } from '../shared/db/executions';
import {
findWorkflowStatistic,
findWorkflowStatistics,
useWorkflowStatisticsEvents,
} from '../shared/workflow-statistics';
describe('ScalingService queue recovery', () => {
let scalingService: ScalingService;
let workflow: WorkflowEntity;
useWorkflowStatisticsEvents();
beforeAll(async () => {
await testDb.init();
Container.get(InstanceSettings).markAsLeader();
Container.get(WorkflowStatisticsService);
scalingService = new ScalingService(
mockLogger(),
mock(),
mock(),
mock(),
Container.get(GlobalConfig),
Container.get(ExecutionRepository),
mock(),
Container.get(InstanceSettings),
mock(),
mock(),
Container.get(ExecutionCrashService),
);
});
beforeEach(async () => {
vi.spyOn(scalingService, 'findJobsByStatus').mockResolvedValue([]);
await testDb.truncate([
'WorkflowEntity',
'WorkflowStatistics',
'WorkflowStatisticsDelta',
'ExecutionEntity',
]);
workflow = await createWorkflow();
});
afterAll(async () => {
await testDb.terminate();
});
const createDanglingExecution = async (
mode: WorkflowExecuteMode,
target: WorkflowEntity = workflow,
) => await createExecution({ status: 'running', finished: false, stoppedAt: null, mode }, target);
const findStatistics = async (name: StatisticsNames, workflowId: string = workflow.id) =>
await findWorkflowStatistic(workflowId, name);
it('records a production error for every execution of the same workflow recovered as crashed in one sweep', async () => {
const first = await createDanglingExecution('trigger');
const second = await createDanglingExecution('trigger');
await scalingService.recoverFromQueue();
const executionRepository = Container.get(ExecutionRepository);
await expect(executionRepository.findOneBy({ id: first.id })).resolves.toMatchObject({
status: 'crashed',
});
await expect(executionRepository.findOneBy({ id: second.id })).resolves.toMatchObject({
status: 'crashed',
});
await vi.waitFor(async () => {
const statistics = await findStatistics(StatisticsNames.productionError);
expect(statistics).toMatchObject({ count: 2, rootCount: 2, workflowName: workflow.name });
});
});
it('does not count a chat execution recovered as crashed', async () => {
const chatWorkflow = await createWorkflow();
await createDanglingExecution('chat', chatWorkflow);
await scalingService.recoverFromQueue();
await createDanglingExecution('trigger');
await scalingService.recoverFromQueue();
// The later sweep's counter appearing proves the chat sweep has finished counting.
await vi.waitFor(async () => {
expect(await findStatistics(StatisticsNames.productionError)).toMatchObject({ count: 1 });
});
expect(await findWorkflowStatistics(chatWorkflow.id)).toEqual([]);
});
it('excludes an execution already `crashed` from a later, sequential sweep', async () => {
const execution = await createDanglingExecution('trigger');
await scalingService.recoverFromQueue();
await vi.waitFor(async () => {
expect(await findStatistics(StatisticsNames.productionError)).toMatchObject({ count: 1 });
});
await scalingService.recoverFromQueue();
await Container.get(ExecutionCrashService).markAsCrashed(execution.id);
const laterWorkflow = await createWorkflow();
await createDanglingExecution('trigger', laterWorkflow);
await scalingService.recoverFromQueue();
// The later counter appearing proves any repeat of the first would already be visible.
await vi.waitFor(async () => {
expect(await findStatistics(StatisticsNames.productionError, laterWorkflow.id)).toMatchObject(
{ count: 1 },
);
});
expect(await findStatistics(StatisticsNames.productionError)).toMatchObject({
count: 1,
rootCount: 1,
});
});
it('attributes each execution in one sweep to its own workflow and mode', async () => {
const manualWorkflow = await createWorkflow();
const subWorkflow = await createWorkflow();
await createDanglingExecution('trigger');
await createDanglingExecution('manual', manualWorkflow);
await createDanglingExecution('integrated', subWorkflow);
await scalingService.recoverFromQueue();
await vi.waitFor(async () => {
expect(await findStatistics(StatisticsNames.productionError)).toMatchObject({
count: 1,
rootCount: 1,
});
expect(await findStatistics(StatisticsNames.manualError, manualWorkflow.id)).toMatchObject({
count: 1,
rootCount: 0,
});
expect(await findStatistics(StatisticsNames.productionError, subWorkflow.id)).toMatchObject({
count: 1,
rootCount: 0,
});
});
expect(await findStatistics(StatisticsNames.manualError)).toBeNull();
expect(await findStatistics(StatisticsNames.productionError, manualWorkflow.id)).toBeNull();
expect(await findStatistics(StatisticsNames.manualError, subWorkflow.id)).toBeNull();
});
});

43
packages/cli/test/integration/shared/workflow-statistics.ts Normal file
View File

@ -0,0 +1,43 @@
import { GlobalConfig } from '@n8n/config';
import { WorkflowStatisticsRepository, type StatisticsNames } from '@n8n/db';
import { Container } from '@n8n/di';
/**
* Fold the pending Postgres deltas, so a read sees the same materialised counter that the
* SQLite path writes directly. No-op on SQLite.
*/
const foldPendingIncrements = async () => {
if (Container.get(GlobalConfig).database.type !== 'postgresdb') return;
const repository = Container.get(WorkflowStatisticsRepository);
await repository.rollupIncrements(repository.manager, 10_000);
};
/**
* Let the statistics listeners run for this test file, which the shared test setup switches
* off, and put the original setting back afterwards. Call it at the top of the `describe`,
* so the hooks it registers run before the file's own.
*/
export const useWorkflowStatisticsEvents = () => {
const original = process.env.SKIP_STATISTICS_EVENTS;
beforeAll(() => {
delete process.env.SKIP_STATISTICS_EVENTS;
});
afterAll(() => {
if (original !== undefined) process.env.SKIP_STATISTICS_EVENTS = original;
});
};
export const findWorkflowStatistic = async (workflowId: string, name: StatisticsNames) => {
await foldPendingIncrements();
return await Container.get(WorkflowStatisticsRepository).findOneBy({ workflowId, name });
};
export const findWorkflowStatistics = async (workflowId: string) => {
await foldPendingIncrements();
return await Container.get(WorkflowStatisticsRepository).findBy({ workflowId });
};