From 1deee723a333aaf9619ec5e0e4d940d9ad29f236 Mon Sep 17 00:00:00 2001 From: Matsu Date: Thu, 6 Aug 2026 16:07:01 +0300 Subject: [PATCH] fix(core): Handle errors from triggered and polled workflow executions (#35689) Co-authored-by: Claude Opus 5 --- .../backend-test-utils/src/db/workflows.ts | 8 +- .../execution-lifecycle-hooks.test.ts | 12 + .../execute-error-workflow.ts | 41 ++- .../execution-lifecycle-hooks.ts | 2 + packages/cli/src/workflow-runner.ts | 2 + .../trigger-execution-context.factory.test.ts | 267 +++++++++++++++++- .../trigger-execution-context.factory.ts | 142 ++++++---- .../cli/test/integration/webhooks.api.test.ts | 6 +- ...er-with-active-workflow-manager.ee.test.ts | 8 + .../test/unit/no-source-import-cycles.test.ts | 150 ++++++++++ 10 files changed, 574 insertions(+), 64 deletions(-) create mode 100644 packages/cli/test/unit/no-source-import-cycles.test.ts diff --git a/packages/@n8n/backend-test-utils/src/db/workflows.ts b/packages/@n8n/backend-test-utils/src/db/workflows.ts index 86f9d1ac08c..06f628da28a 100644 --- a/packages/@n8n/backend-test-utils/src/db/workflows.ts +++ b/packages/@n8n/backend-test-utils/src/db/workflows.ts @@ -191,7 +191,13 @@ export async function createWorkflowWithTrigger( }, { id: 'uuid-2', - parameters: { triggerTimes: { item: [{ mode: 'everyMinute' }] } }, + // Deliberately a schedule that cannot fire during a test run. Suites + // that don't mock `ActiveWorkflowManager` register this with the real + // `ScheduledTaskManager`, and an `everyMinute` tick starts a real + // execution that outlives the test that activated the workflow. + parameters: { + triggerTimes: { item: [{ mode: 'everyMonth', hour: 0, minute: 0, dayOfMonth: 1 }] }, + }, name: 'Cron', type: 'n8n-nodes-base.cron', typeVersion: 1, diff --git a/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts b/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts index 626ea69e9c5..f3b6d151c10 100644 --- a/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts +++ b/packages/cli/src/execution-lifecycle/__tests__/execution-lifecycle-hooks.test.ts @@ -63,6 +63,14 @@ describe('Execution Lifecycle Hooks', () => { const redactionProxy = mockInstance(ExecutionRedactionServiceProxy); const workflowHookContext = mockInstance(WorkflowHookContextService); + /** + * The error-workflow dispatch is deliberately fire-and-forget: the hook does + * not await it, and it resolves `WorkflowExecutionService` through a lazy + * import. Drain the microtask queue (this suite fakes timers, so `nextTick` + * rather than `setImmediate`) before asserting on it. + */ + const flushErrorWorkflowDispatch = async () => await new Promise(process.nextTick); + const nodeName = 'Test Node'; const nodeType = 'n8n-nodes-base.testNode'; const nodeId = 'test-node-id'; @@ -1121,6 +1129,7 @@ describe('Execution Lifecycle Hooks', () => { await lifecycleHooks.runHook('workflowExecuteAfter', [failedRun, {}]); + await flushErrorWorkflowDispatch(); expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalledWith( errorWorkflow, { @@ -1178,6 +1187,7 @@ describe('Execution Lifecycle Hooks', () => { await lifecycleHooks.runHook('workflowExecuteAfter', [failedRun, {}]); + await flushErrorWorkflowDispatch(); expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalledWith( errorWorkflow, expect.objectContaining({ workflow: { id: workflowId, name: workflowData.name } }), @@ -1542,6 +1552,7 @@ describe('Execution Lifecycle Hooks', () => { await lifecycleHooks.runHook('workflowExecuteAfter', [failedRun, {}]); + await flushErrorWorkflowDispatch(); expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalledWith( errorWorkflow, { @@ -1596,6 +1607,7 @@ describe('Execution Lifecycle Hooks', () => { await lifecycleHooks.runHook('workflowExecuteAfter', [failedRun, {}]); + await flushErrorWorkflowDispatch(); expect(workflowExecutionService.executeErrorWorkflow).toHaveBeenCalledWith( errorWorkflow, expect.objectContaining({ workflow: { id: workflowId, name: workflowData.name } }), diff --git a/packages/cli/src/execution-lifecycle/execute-error-workflow.ts b/packages/cli/src/execution-lifecycle/execute-error-workflow.ts index f89f19fabcb..de2fc4c3bc7 100644 --- a/packages/cli/src/execution-lifecycle/execute-error-workflow.ts +++ b/packages/cli/src/execution-lifecycle/execute-error-workflow.ts @@ -7,8 +7,26 @@ import type { IRun, IWorkflowBase, WorkflowExecuteMode } from 'n8n-workflow'; import type { IWorkflowErrorData } from '@/interfaces'; import { OwnershipService } from '@/services/ownership.service'; import { UrlService } from '@/services/url.service'; -// eslint-disable-next-line import-x/no-cycle -import { WorkflowExecutionService } from '@/workflows/workflow-execution.service'; + +/** + * Resolved lazily. A static import would close the cycle + * `workflow-runner -> execution-lifecycle-hooks -> execute-error-workflow -> + * workflow-execution.service -> workflow-runner`. Enter that graph at the wrong + * point (`workflow-runner`, `wait-tracker`, `webhook-helpers` or + * `test-webhooks`) and `emitDecoratorMetadata` captures an unresolved paramtype + * for `WorkflowExecutionService`'s `workflowRunner`, which DI silently injects + * as `undefined` — the next triggered execution then dies with "Cannot read + * properties of undefined (reading 'run')". + */ +const getWorkflowExecutionService = async () => { + // `no-cycle` counts a dynamic import as an edge. It isn't one that matters + // here: the import is evaluated at call time, long after every module in the + // chain is initialised, so it creates no evaluation-order dependency. That is + // the whole reason this is a dynamic import. + // eslint-disable-next-line import-x/no-cycle + const { WorkflowExecutionService } = await import('@/workflows/workflow-execution.service.js'); + return Container.get(WorkflowExecutionService); +}; /** * Checks if there was an error and if errorWorkflow or a trigger is defined. If so it collects @@ -93,8 +111,9 @@ export function executeErrorWorkflow( Container.get(OwnershipService) .getWorkflowProjectCached(workflowId) - .then((project) => { - void Container.get(WorkflowExecutionService).executeErrorWorkflow( + .then(async (project) => { + const workflowExecutionService = await getWorkflowExecutionService(); + await workflowExecutionService.executeErrorWorkflow( errorWorkflow, workflowErrorData, project, @@ -121,12 +140,22 @@ export function executeErrorWorkflow( logger.debug('Start internal error workflow', { executionId, workflowId }); void Container.get(OwnershipService) .getWorkflowProjectCached(workflowId) - .then((project) => { - void Container.get(WorkflowExecutionService).executeErrorWorkflow( + .then(async (project) => { + const workflowExecutionService = await getWorkflowExecutionService(); + await workflowExecutionService.executeErrorWorkflow( workflowId, workflowErrorData, project, ); + }) + .catch((error: Error) => { + Container.get(ErrorReporter).error(error); + logger.error(`Could not execute internal ErrorWorkflow for execution ID ${executionId}`, { + executionId, + workflowId, + error, + workflowErrorData, + }); }); } } diff --git a/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts b/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts index 5b10ca80387..b662ca644ab 100644 --- a/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts +++ b/packages/cli/src/execution-lifecycle/execution-lifecycle-hooks.ts @@ -36,6 +36,8 @@ import { getLastExecutedNodeData } from '@/workflow-helpers'; import { WorkflowHookContextService } from '@/workflow-hook-context.service'; import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service'; +// `no-cycle` still reports a cycle here, but only through the dynamic import +// in `execute-error-workflow`, which creates no evaluation-order edge. // eslint-disable-next-line import-x/no-cycle import { executeErrorWorkflow } from './execute-error-workflow'; import { restoreBinaryDataId } from './restore-binary-data-id'; diff --git a/packages/cli/src/workflow-runner.ts b/packages/cli/src/workflow-runner.ts index ac5f1a0b43a..5deec9651f6 100644 --- a/packages/cli/src/workflow-runner.ts +++ b/packages/cli/src/workflow-runner.ts @@ -35,6 +35,8 @@ import PCancelable from 'p-cancelable'; import { ActiveExecutions } from '@/active-executions'; import { ExecutionNotFoundError } from '@/errors/execution-not-found-error'; import { MaxStalledCountError } from '@/errors/max-stalled-count.error'; +// `no-cycle` still reports a cycle here, but only through the dynamic import +// in `execute-error-workflow`, which creates no evaluation-order edge. // eslint-disable-next-line import-x/no-cycle import { getLifecycleHooksForRegularMain, diff --git a/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts b/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts index 944711c24cf..42fef548803 100644 --- a/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts @@ -1,6 +1,7 @@ /* eslint-disable @typescript-eslint/unbound-method */ import type { Logger } from '@n8n/backend-common'; import type { Project, WorkflowEntity } from '@n8n/db'; +import type { IDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import { sleep } from '@n8n/utils/sleep'; import type { ErrorReporter, IGetExecutePollFunctions, StorageConfig } from 'n8n-core'; @@ -61,6 +62,8 @@ describe('TriggerExecutionContextFactory', () => { const nodeTypes = createNodeTypes(); let factory: TriggerExecutionContextFactory; + let logger: Logger; + let errorReporter: ErrorReporter; beforeEach(() => { vi.clearAllMocks(); @@ -72,12 +75,13 @@ describe('TriggerExecutionContextFactory', () => { ); scheduleTriggerJobRegistrar.interceptsNode.mockReturnValue(false); - const scopedLogger = mock(); - const rootLogger = mock({ scoped: vi.fn().mockReturnValue(scopedLogger) }); + logger = mock(); + const rootLogger = mock({ scoped: vi.fn().mockReturnValue(logger) }); + errorReporter = mock(); factory = new TriggerExecutionContextFactory( rootLogger, - mock(), + errorReporter, activeExecutions, eventService, executionService, @@ -255,6 +259,171 @@ describe('TriggerExecutionContextFactory', () => { await expect(donePromise.promise).resolves.toBeUndefined(); }); + + describe('when the execution fails', () => { + // A trigger fires on a timer, so nothing is awaiting `emit`. Any promise + // it derives without a terminal handler surfaces as an unhandled + // rejection, which fails the whole vitest run even though every + // assertion passed. See DEVP-687. + const unhandled: unknown[] = []; + const captureUnhandled = (reason: unknown) => unhandled.push(reason); + + beforeEach(() => { + unhandled.length = 0; + process.on('unhandledRejection', captureUnhandled); + }); + + afterEach(() => { + process.off('unhandledRejection', captureUnhandled); + }); + + /** Let Node reach the checkpoint where it reports unhandled rejections. */ + const flush = async () => { + await sleep(0); + await new Promise((resolve) => setImmediate(resolve)); + }; + + const emitWith = (donePromise?: IDeferredPromise) => { + const workflowData = mock({ id: 'wf-1', name: 'Test Workflow' }); + const additionalData = mock(); + const mode: WorkflowExecuteMode = 'trigger'; + const activation: WorkflowActivateMode = 'activate'; + const workflow = mock({ name: 'Test Workflow' }); + const node = mock({ name: 'Trigger Node', id: 'node-1' }); + + const getTriggerFunctions = factory.getExecuteTriggerFunctions( + workflowData, + additionalData, + mode, + activation, + async () => workflowData, + vi.fn(), + scheduleCollectionSession, + ); + const context = getTriggerFunctions(workflow, node, additionalData, mode, activation); + + context.emit([[]], undefined, donePromise); + }; + + test('logs the error and does not emit workflow-executed', async () => { + workflowExecutionService.runWorkflow.mockRejectedValueOnce(new UnexpectedError('boom')); + + emitWith(); + await flush(); + + expect(unhandled).toEqual([]); + expect(logger.error).toHaveBeenCalledWith('boom', expect.objectContaining({})); + expect(eventService.emit).not.toHaveBeenCalled(); + }); + + test('rejects donePromise instead of leaving it pending', async () => { + workflowExecutionService.runWorkflow.mockRejectedValueOnce(new UnexpectedError('boom')); + const donePromise = createDeferredPromise(); + + emitWith(donePromise); + + await expect(donePromise.promise).rejects.toThrow('boom'); + await flush(); + expect(unhandled).toEqual([]); + }); + + test('rejects donePromise when the workflow data cannot be resolved', async () => { + const workflowData = mock({ id: 'wf-1', name: 'Test Workflow' }); + const donePromise = createDeferredPromise(); + + const getTriggerFunctions = factory.getExecuteTriggerFunctions( + workflowData, + mock(), + 'trigger', + 'activate', + async () => { + throw new UnexpectedError('Published version not found for workflow'); + }, + vi.fn(), + scheduleCollectionSession, + ); + const context = getTriggerFunctions( + mock({ name: 'Test Workflow' }), + mock({ name: 'Trigger Node' }), + mock(), + 'trigger', + 'activate', + ); + + context.emit([[]], undefined, donePromise); + + await expect(donePromise.promise).rejects.toThrow('Published version not found'); + await flush(); + expect(unhandled).toEqual([]); + expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled(); + }); + + test('rejects donePromise with an Error when the failure is not one', async () => { + workflowExecutionService.runWorkflow.mockRejectedValueOnce('just a string'); + const donePromise = createDeferredPromise(); + + emitWith(donePromise); + + await expect(donePromise.promise).rejects.toBeInstanceOf(Error); + await flush(); + expect(unhandled).toEqual([]); + }); + + test('rejects donePromise when the post-execute promise fails', async () => { + activeExecutions.getPostExecutePromise.mockRejectedValueOnce( + new UnexpectedError('execution gone'), + ); + const donePromise = createDeferredPromise(); + + emitWith(donePromise); + + await expect(donePromise.promise).rejects.toThrow('execution gone'); + await flush(); + expect(unhandled).toEqual([]); + }); + + test('does not report an unhandled rejection on the happy path', async () => { + activeExecutions.getPostExecutePromise.mockResolvedValue(mock()); + const donePromise = createDeferredPromise(); + + emitWith(donePromise); + await flush(); + + expect(unhandled).toEqual([]); + expect(eventService.emit).toHaveBeenCalledTimes(1); + }); + + test('logs when the failed execution cannot be recorded', async () => { + vi.spyOn(factory, 'executeErrorWorkflow').mockImplementation(() => {}); + executionService.createErrorExecution.mockRejectedValueOnce( + new UnexpectedError('db down'), + ); + + const workflowData = mock({ id: 'wf-1', name: 'Test Workflow' }); + const getTriggerFunctions = factory.getExecuteTriggerFunctions( + workflowData, + mock(), + 'trigger', + 'activate', + async () => workflowData, + vi.fn(), + scheduleCollectionSession, + ); + const context = getTriggerFunctions( + mock({ name: 'Test Workflow' }), + mock({ name: 'Trigger Node' }), + mock(), + 'trigger', + 'activate', + ); + + context.saveFailedExecution(mock()); + await flush(); + + expect(unhandled).toEqual([]); + expect(errorReporter.error).toHaveBeenCalled(); + }); + }); }); describe('emitError', () => { @@ -455,6 +624,98 @@ describe('TriggerExecutionContextFactory', () => { expect(activeExecutions.getPostExecutePromise).toHaveBeenCalledWith('exec-123'); await expect(donePromise.promise).resolves.toBe(runResult); }); + + describe('when the execution fails', () => { + // See the equivalent block for `emit` — a poll fires on a timer too. + const unhandled: unknown[] = []; + const captureUnhandled = (reason: unknown) => unhandled.push(reason); + + beforeEach(() => { + unhandled.length = 0; + process.on('unhandledRejection', captureUnhandled); + }); + + afterEach(() => { + process.off('unhandledRejection', captureUnhandled); + }); + + const flush = async () => { + await sleep(0); + await new Promise((resolve) => setImmediate(resolve)); + }; + + const pollEmitWith = (donePromise?: IDeferredPromise) => { + const workflowData = mock({ id: 'wf-1', name: 'Test Workflow' }); + const additionalData = mock(); + + const getPollFunctions = factory.getExecutePollFunctions( + workflowData, + additionalData, + 'trigger', + 'activate', + async () => workflowData, + ); + const context = getPollFunctions( + mock({ id: 'wf-1', name: 'Test Workflow' }), + mock({ name: 'Poll Node' }), + additionalData, + 'trigger', + 'activate', + ); + + context.__emit([[]], undefined, donePromise); + }; + + test('logs the error without a donePromise', async () => { + workflowExecutionService.runWorkflow.mockRejectedValueOnce(new UnexpectedError('boom')); + + pollEmitWith(); + await flush(); + + expect(unhandled).toEqual([]); + expect(logger.error).toHaveBeenCalledWith('boom', expect.objectContaining({})); + }); + + test('rejects donePromise instead of leaving it pending', async () => { + workflowExecutionService.runWorkflow.mockRejectedValueOnce(new UnexpectedError('boom')); + const donePromise = createDeferredPromise(); + + pollEmitWith(donePromise); + + await expect(donePromise.promise).rejects.toThrow('boom'); + await flush(); + expect(unhandled).toEqual([]); + }); + + test('logs when the failed execution cannot be recorded', async () => { + vi.spyOn(factory, 'executeErrorWorkflow').mockImplementation(() => {}); + executionService.createErrorExecution.mockRejectedValueOnce( + new UnexpectedError('db down'), + ); + + const workflowData = mock({ id: 'wf-1', name: 'Test Workflow' }); + const getPollFunctions = factory.getExecutePollFunctions( + workflowData, + mock(), + 'trigger', + 'activate', + async () => workflowData, + ); + const context = getPollFunctions( + mock({ id: 'wf-1', name: 'Test Workflow' }), + mock({ name: 'Poll Node' }), + mock(), + 'trigger', + 'activate', + ); + + context.__emitError(mock()); + await flush(); + + expect(unhandled).toEqual([]); + expect(errorReporter.error).toHaveBeenCalled(); + }); + }); }); describe('__emitError', () => { diff --git a/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts b/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts index 570220b112d..e47fb4c9bdb 100644 --- a/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts +++ b/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts @@ -1,6 +1,7 @@ import { Logger } from '@n8n/backend-common'; import type { IWorkflowDb } 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'; import { ErrorReporter, @@ -76,6 +77,65 @@ export class TriggerExecutionContextFactory { this.logger = this.logger.scoped(['workflow-activation']); } + /** + * Bridge an in-flight triggered execution to the node's `donePromise`. It + * always settles: with the finished run, with `undefined` when a duplicate + * scheduled execution was skipped, or rejected if the execution never + * started. Nodes that await it (MQTT, RabbitMQ, AMQP, Kafka) would otherwise + * hang forever on failure. + */ + private settleDonePromise( + executePromise: Promise, + donePromise: IDeferredPromise, + ) { + void executePromise + .then(async (executionId) => + executionId === undefined + ? undefined + : await this.activeExecutions.getPostExecutePromise(executionId), + ) + .then(donePromise.resolve, (error: unknown) => donePromise.reject(ensureError(error))); + } + + /** + * Terminates the emit promise chains. Without it a failed triggered execution + * surfaces as an unhandled rejection instead of a logged error. + */ + private logTriggerExecutionFailure(error: unknown, workflowData: IWorkflowBase, node: INode) { + const failure = ensureError(error); + this.logger.error(failure.message, { + error: failure, + workflowId: workflowData.id, + nodeName: node.name, + }); + } + + /** + * Persist a failed trigger execution, then run the error workflow for it. + */ + private recordTriggerFailure( + error: ExecutionError, + node: INode, + workflowData: IWorkflowBase, + workflow: Workflow, + mode: WorkflowExecuteMode, + ) { + void this.executionService + .createErrorExecution(error, node, workflowData, workflow, mode) + .then(() => { + this.executeErrorWorkflow(error, workflowData, mode); + }) + .catch((cause: unknown) => { + const failure = ensureError(cause); + this.errorReporter.error(failure); + this.logger.error('Failed to record failed trigger execution', { + error: failure, + workflowId: workflowData.id, + nodeName: node.name, + }); + }); + } + /** * Return trigger function which gets the global functions from n8n-core * and overwrites the emit to be able to start it in subprocess @@ -136,40 +196,29 @@ export class TriggerExecutionContextFactory { throw error; }); - void executePromise.then(async (executionId) => { - // `executionId` is undefined when the catch above swallowed a - // duplicate scheduled execution; nothing ran, so nothing to emit. - if (executionId === undefined) return; - const { projectId, projectName } = await getWorkflowProjectDetailsSafe( - this.ownershipService, - workflowData.id, - ); - this.eventService.emit('workflow-executed', { - workflowId: workflowData.id, - workflowName: workflowData.name, - executionId, - projectId, - projectName, - source: 'trigger', - }); - }); + // Registered ahead of the telemetry chain so the post-execute promise + // is attached before that chain's ownership lookup yields. + if (donePromise) this.settleDonePromise(executePromise, donePromise); - if (donePromise) { - void executePromise.then((executionId) => { - // Same as above: a duplicate scheduled execution was skipped, - // so resolve with undefined and don't wait on a non-existent run. - if (executionId === undefined) { - donePromise.resolve(undefined); - return; - } - this.activeExecutions - .getPostExecutePromise(executionId) - .then(donePromise.resolve) - .catch(donePromise.reject); - }); - } else { - executePromise.catch((error: Error) => this.logger.error(error.message, { error })); - } + void executePromise + .then(async (executionId) => { + // `executionId` is undefined when the catch above swallowed a + // duplicate scheduled execution; nothing ran, so nothing to emit. + if (executionId === undefined) return; + const { projectId, projectName } = await getWorkflowProjectDetailsSafe( + this.ownershipService, + workflowData.id, + ); + this.eventService.emit('workflow-executed', { + workflowId: workflowData.id, + workflowName: workflowData.name, + executionId, + projectId, + projectName, + source: 'trigger', + }); + }) + .catch((error: unknown) => this.logTriggerExecutionFailure(error, workflowData, node)); }; const emitError = (error: Error): void => { @@ -185,11 +234,7 @@ export class TriggerExecutionContextFactory { workflowName: workflowData.name, }, ); - void this.executionService - .createErrorExecution(error, node, workflowData, workflow, mode) - .then(() => { - this.executeErrorWorkflow(error, workflowData, mode); - }); + this.recordTriggerFailure(error, node, workflowData, workflow, mode); }; const schedulingFunctions = this.scheduleTriggerJobRegistrar.interceptsNode(node) @@ -249,24 +294,15 @@ export class TriggerExecutionContextFactory { ), ); - if (donePromise) { - void executePromise.then((executionId) => { - this.activeExecutions - .getPostExecutePromise(executionId) - .then(donePromise.resolve) - .catch(donePromise.reject); - }); - } else { - void executePromise.catch((error: Error) => this.logger.error(error.message, { error })); - } + if (donePromise) this.settleDonePromise(executePromise, donePromise); + + void executePromise.catch((error: unknown) => + this.logTriggerExecutionFailure(error, workflowData, node), + ); }; const __emitError = (error: ExecutionError) => { - void this.executionService - .createErrorExecution(error, node, workflowData, workflow, mode) - .then(() => { - this.executeErrorWorkflow(error, workflowData, mode); - }); + this.recordTriggerFailure(error, node, workflowData, workflow, mode); }; return new PollContext(workflow, node, additionalData, mode, activation, __emit, __emitError); diff --git a/packages/cli/test/integration/webhooks.api.test.ts b/packages/cli/test/integration/webhooks.api.test.ts index 5a88be71798..7f52c9a20f9 100644 --- a/packages/cli/test/integration/webhooks.api.test.ts +++ b/packages/cli/test/integration/webhooks.api.test.ts @@ -87,6 +87,7 @@ describe('Webhook API', () => { let user: User; let agent: SuperAgentTest; let workflow: WorkflowEntity | undefined; + let activeWorkflowManager: Awaited> | undefined; beforeAll(async () => { await testDb.init(); @@ -100,10 +101,13 @@ describe('Webhook API', () => { beforeEach(async () => { await testDb.truncate(['WorkflowEntity']); workflow = await createActiveWorkflow(workflowData, user); - await initActiveWorkflowManager(); + activeWorkflowManager = await initActiveWorkflowManager(); }); afterEach(async () => { + // The manager is re-inited per test, so without this each run leaves its + // registrations live for the rest of the worker. + await activeWorkflowManager?.removeAll(); if (workflow) { await deleteWorkflowAndWebhooks(workflow.id); } diff --git a/packages/cli/test/integration/workflows/workflows.controller-with-active-workflow-manager.ee.test.ts b/packages/cli/test/integration/workflows/workflows.controller-with-active-workflow-manager.ee.test.ts index 804fc8cb271..11b0eb02444 100644 --- a/packages/cli/test/integration/workflows/workflows.controller-with-active-workflow-manager.ee.test.ts +++ b/packages/cli/test/integration/workflows/workflows.controller-with-active-workflow-manager.ee.test.ts @@ -5,7 +5,9 @@ import { createActiveWorkflow, } from '@n8n/backend-test-utils'; import type { User } from '@n8n/db'; +import { Container } from '@n8n/di'; +import { ActiveWorkflowManager } from '@/active-workflow-manager'; import { Telemetry } from '@/telemetry'; import { createUser } from '../shared/db/users'; @@ -35,6 +37,12 @@ beforeEach(async () => { ]); }); +afterEach(async () => { + // This suite drives the real ActiveWorkflowManager, so anything it registers + // (cron jobs, webhooks) would otherwise stay live for the rest of the worker. + await Container.get(ActiveWorkflowManager).removeAll(); +}); + describe('PUT /:workflowId/transfer', () => { // This tests does not mock the ActiveWorkflowManager, which helps catching // possible deadlocks when using transactions wrong. diff --git a/packages/cli/test/unit/no-source-import-cycles.test.ts b/packages/cli/test/unit/no-source-import-cycles.test.ts new file mode 100644 index 00000000000..306ba9dc12f --- /dev/null +++ b/packages/cli/test/unit/no-source-import-cycles.test.ts @@ -0,0 +1,150 @@ +import { readdirSync, readFileSync, statSync } from 'node:fs'; +import path from 'node:path'; + +/** + * A value-import cycle between two `@Service()` modules is not a style problem: + * whichever module the graph is entered at gets an unresolved `design:paramtypes` + * entry for the other, and `@n8n/di` silently injects `undefined` for it. The + * class then fails at call time with "Cannot read properties of undefined", + * hundreds of test files away from the import that caused it. + * + * `import-x/no-cycle` is only a warning, and it reports per-import rather than + * per-cycle, so this test is the gate. + */ + +const SRC = path.resolve(__dirname, '../../src'); + +const resolveSpecifier = (specifier: string, importer: string): string | null => { + let base: string; + if (specifier.startsWith('@/')) base = path.join(SRC, specifier.slice(2)); + else if (specifier.startsWith('.')) base = path.resolve(path.dirname(importer), specifier); + else return null; + + // `@/foo.js` is the on-disk `@/foo.ts` (NodeNext-style specifiers). + const withoutJs = base.endsWith('.js') ? base.slice(0, -3) : base; + + for (const candidate of [withoutJs + '.ts', path.join(withoutJs, 'index.ts')]) { + try { + if (statSync(candidate).isFile()) return candidate; + } catch { + // not this candidate + } + } + return null; +}; + +/** + * Static value imports and re-exports only. `import type` and dynamic + * `await import()` do not create an evaluation-order edge, which is exactly + * why they are the escape hatch when a cycle is unavoidable. + */ +// Clauses never contain quotes or semicolons, so [^'";]*? spans multi-line +// braced imports without leaking past a bare side-effect import's statement. +const importPattern = /(?:^|\n)\s*import\s+(type\s+)?(?:[^'";]*?\bfrom\s*)?['"]([^'"]+)['"]/g; +// Re-exports (`export ... from`) are evaluation-order edges too — barrels route cycles. +const exportPattern = + /(?:^|\n)\s*export\s+(type\s+)?(?:\*(?:\s+as\s+[\w$]+)?|\{[^}]*\})\s*from\s*['"]([^'"]+)['"]/g; + +const valueImportsOf = (file: string): string[] => { + const source = readFileSync(file, 'utf8'); + const edges: string[] = []; + + for (const pattern of [importPattern, exportPattern]) { + for (const match of source.matchAll(pattern)) { + if (match[1]) continue; + const resolved = resolveSpecifier(match[2], file); + if (resolved) edges.push(resolved); + } + } + return edges; +}; + +/** Tarjan's algorithm, iterative to stay clear of the call-stack limit. */ +const findCycles = (nodes: string[], edgesOf: (node: string) => string[]): string[][] => { + const index = new Map(); + const lowlink = new Map(); + const onStack = new Set(); + const stack: string[] = []; + const cycles: string[][] = []; + let counter = 0; + + for (const root of nodes) { + if (index.has(root)) continue; + + const work: Array<{ node: string; edge: number }> = [{ node: root, edge: 0 }]; + index.set(root, counter); + lowlink.set(root, counter); + counter += 1; + stack.push(root); + onStack.add(root); + + while (work.length > 0) { + const frame = work[work.length - 1]; + const neighbours = edgesOf(frame.node); + + if (frame.edge < neighbours.length) { + const next = neighbours[frame.edge]; + frame.edge += 1; + + if (!index.has(next)) { + index.set(next, counter); + lowlink.set(next, counter); + counter += 1; + stack.push(next); + onStack.add(next); + work.push({ node: next, edge: 0 }); + } else if (onStack.has(next)) { + lowlink.set(frame.node, Math.min(lowlink.get(frame.node)!, index.get(next)!)); + } + continue; + } + + work.pop(); + const parent = work[work.length - 1]; + if (parent) { + lowlink.set(parent.node, Math.min(lowlink.get(parent.node)!, lowlink.get(frame.node)!)); + } + + if (lowlink.get(frame.node) === index.get(frame.node)) { + const component: string[] = []; + let member: string; + do { + member = stack.pop()!; + onStack.delete(member); + component.push(member); + } while (member !== frame.node); + + if (component.length > 1) cycles.push(component); + } + } + } + + return cycles; +}; + +test('packages/cli/src has no value-import cycles', () => { + const files = readdirSync(SRC, { recursive: true, encoding: 'utf8' }) + .filter( + (file) => file.endsWith('.ts') && !file.includes('__tests__') && !file.endsWith('.test.ts'), + ) + .map((file) => path.join(SRC, file)); + + const cache = new Map(); + const edgesOf = (file: string) => { + let edges = cache.get(file); + if (!edges) { + edges = valueImportsOf(file); + cache.set(file, edges); + } + return edges; + }; + + const cycles = findCycles(files, edgesOf) + // Only a cycle that reaches a DI service can corrupt an injected + // dependency. TypeORM entities, for one, reference each other by design + // for bidirectional relations and have no injected constructor params. + .filter((cycle) => cycle.some((file) => readFileSync(file, 'utf8').includes('@Service('))) + .map((cycle) => cycle.map((file) => path.relative(SRC, file)).sort()); + + expect(cycles).toEqual([]); +});