diff --git a/packages/@n8n/db/src/repositories/__tests__/execution.repository.test.ts b/packages/@n8n/db/src/repositories/__tests__/execution.repository.test.ts index 42cb3c3752a..5701203d96f 100644 --- a/packages/@n8n/db/src/repositories/__tests__/execution.repository.test.ts +++ b/packages/@n8n/db/src/repositories/__tests__/execution.repository.test.ts @@ -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(); + 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({ + id, + workflowId: `workflow-${id}`, + mode: 'trigger', + workflow: mock({ 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', () => { diff --git a/packages/@n8n/db/src/repositories/execution.repository.ts b/packages/@n8n/db/src/repositories/execution.repository.ts index 82000fe9a40..acd1e9e2037 100644 --- a/packages/@n8n/db/src/repositories/execution.repository.ts +++ b/packages/@n8n/db/src/repositories/execution.repository.ts @@ -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 { } 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 { + // 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 { + 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, + ): Promise { + // `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) { diff --git a/packages/@n8n/db/src/repositories/index.ts b/packages/@n8n/db/src/repositories/index.ts index dc35e8f6678..3966dcce46a 100644 --- a/packages/@n8n/db/src/repositories/index.ts +++ b/packages/@n8n/db/src/repositories/index.ts @@ -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'; diff --git a/packages/cli/src/eventbus/message-event-bus/__tests__/message-event-bus.test.ts b/packages/cli/src/eventbus/message-event-bus/__tests__/message-event-bus.test.ts index 51d7a3dc8f4..ba71e1ac0cd 100644 --- a/packages/cli/src/eventbus/message-event-bus/__tests__/message-event-bus.test.ts +++ b/packages/cli/src/eventbus/message-event-bus/__tests__/message-event-bus.test.ts @@ -50,6 +50,7 @@ describe('MessageEventBus.initialize', () => { executionRepository, mock(), mock(), + mock(), globalConfig, mock({ n8nFolder: tempDir }), ); diff --git a/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts b/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts index d1f82935a0e..7200a25d3a6 100644 --- a/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts +++ b/packages/cli/src/eventbus/message-event-bus/message-event-bus.ts @@ -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 diff --git a/packages/cli/src/executions/__tests__/enqueued-execution-recovery.service.test.ts b/packages/cli/src/executions/__tests__/enqueued-execution-recovery.service.test.ts index fc98ebf69a1..da599208c43 100644 --- a/packages/cli/src/executions/__tests__/enqueued-execution-recovery.service.test.ts +++ b/packages/cli/src/executions/__tests__/enqueued-execution-recovery.service.test.ts @@ -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(); const executionService = mock(); - const executionRepository = mock(); + const executionCrashService = mock(); const ownershipService = mock(); const workflowRunner = mock(); const eventService = mock(); @@ -32,7 +33,7 @@ describe('EnqueuedExecutionRecoveryService', () => { errorReporter, mock({ 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, diff --git a/packages/cli/src/executions/__tests__/execution-crash.service.test.ts b/packages/cli/src/executions/__tests__/execution-crash.service.test.ts new file mode 100644 index 00000000000..ee3c7727fca --- /dev/null +++ b/packages/cli/src/executions/__tests__/execution-crash.service.test.ts @@ -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(); + const workflowStatisticsService = mock(); + + const crashService = new ExecutionCrashService(executionRepository, workflowStatisticsService); + + const crashedExecution = (id: string, overrides: Partial = {}) => ({ + 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(); + }); +}); diff --git a/packages/cli/src/executions/__tests__/execution-recovery.service.integration.test.ts b/packages/cli/src/executions/__tests__/execution-recovery.service.integration.test.ts index 15c3bb666ab..4f974d4d847 100644 --- a/packages/cli/src/executions/__tests__/execution-recovery.service.integration.test.ts +++ b/packages/cli/src/executions/__tests__/execution-recovery.service.integration.test.ts @@ -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()), ); }); diff --git a/packages/cli/src/executions/enqueued-execution-recovery.service.ts b/packages/cli/src/executions/enqueued-execution-recovery.service.ts index f0d62123387..0b4dea17935 100644 --- a/packages/cli/src/executions/enqueued-execution-recovery.service.ts +++ b/packages/cli/src/executions/enqueued-execution-recovery.service.ts @@ -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); } } diff --git a/packages/cli/src/executions/execution-crash.service.ts b/packages/cli/src/executions/execution-crash.service.ts new file mode 100644 index 00000000000..93fd8143467 --- /dev/null +++ b/packages/cli/src/executions/execution-crash.service.ts @@ -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 { + 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 { + await this.executionRepository.markAsCrashed(executionIds); + } + + async markWorkflowExecutionsAsCrashed(workflowId: string): Promise { + 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 }); + } +} diff --git a/packages/cli/src/executions/execution-recovery.service.ts b/packages/cli/src/executions/execution-recovery.service.ts index 8011f22d9bf..80537d7011d 100644 --- a/packages/cli/src/executions/execution-recovery.service.ts +++ b/packages/cli/src/executions/execution-recovery.service.ts @@ -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) { @@ -106,10 +102,7 @@ export class ExecutionRecoveryService { } } - await this.executionRepository.update( - { workflowId, status: In(['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, diff --git a/packages/cli/src/scaling/__tests__/scaling.service.test.ts b/packages/cli/src/scaling/__tests__/scaling.service.test.ts index 8891f14833d..c6f520b396f 100644 --- a/packages/cli/src/scaling/__tests__/scaling.service.test.ts +++ b/packages/cli/src/scaling/__tests__/scaling.service.test.ts @@ -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(); const executionRepository = mock(); const executionPersistence = mock(); + const executionCrashService = mockInstance(ExecutionCrashService); const webhookResponseRelay = mock(); 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(); }); }); diff --git a/packages/cli/src/scaling/scaling.service.ts b/packages/cli/src/scaling/scaling.service.ts index 33fb158da2a..b089355e7ae 100644 --- a/packages/cli/src/scaling/scaling.service.ts +++ b/packages/cli/src/scaling/scaling.service.ts @@ -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, diff --git a/packages/cli/src/services/workflow-statistics.service.ts b/packages/cli/src/services/workflow-statistics.service.ts index 65b28ab6430..62371c8bd61 100644 --- a/packages/cli/src/services/workflow-statistics.service.ts +++ b/packages/cli/src/services/workflow-statistics.service.ts @@ -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; +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 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 { + // 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 { let upsertResult: Awaited>; try { @@ -146,12 +210,7 @@ export class WorkflowStatisticsService extends TypedEmitter { const nodeTypes = mock(); const workflowRunner = mock(); const pollCursorService = mock(); - const executionRepository = mock(); + const executionCrashService = mock(); const logger = mock(); const errorReporter = mock(); const workflowExecutionService = new WorkflowExecutionService( @@ -127,7 +127,7 @@ describe('WorkflowExecutionService', () => { mock(), mock(), pollCursorService, - executionRepository, + executionCrashService, ); const additionalData = mock({}); @@ -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(); }); diff --git a/packages/cli/src/workflows/workflow-execution.service.ts b/packages/cli/src/workflows/workflow-execution.service.ts index c07cc383e6a..38cde6232b5 100644 --- a/packages/cli/src/workflows/workflow-execution.service.ts +++ b/packages/cli/src/workflows/workflow-execution.service.ts @@ -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) { diff --git a/packages/cli/test/integration/database/repositories/execution.repository.test.ts b/packages/cli/test/integration/database/repositories/execution.repository.test.ts index 26b9937ebb4..61f59c31c20 100644 --- a/packages/cli/test/integration/database/repositories/execution.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/execution.repository.test.ts @@ -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', () => { diff --git a/packages/cli/test/integration/executions/execution-recovery-statistics.test.ts b/packages/cli/test/integration/executions/execution-recovery-statistics.test.ts new file mode 100644 index 00000000000..8bc23f5037f --- /dev/null +++ b/packages/cli/test/integration/executions/execution-recovery-statistics.test.ts @@ -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 }); + }); + }); +}); diff --git a/packages/cli/test/integration/scaling/queue-recovery.test.ts b/packages/cli/test/integration/scaling/queue-recovery.test.ts new file mode 100644 index 00000000000..02fd24b00fc --- /dev/null +++ b/packages/cli/test/integration/scaling/queue-recovery.test.ts @@ -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(); + }); +}); diff --git a/packages/cli/test/integration/shared/workflow-statistics.ts b/packages/cli/test/integration/shared/workflow-statistics.ts new file mode 100644 index 00000000000..7b8da2fd6ef --- /dev/null +++ b/packages/cli/test/integration/shared/workflow-statistics.ts @@ -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 }); +};