From faf4dc35dfd1e855fb3554d7445e8836e98fd456 Mon Sep 17 00:00:00 2001 From: Danny Martini Date: Mon, 17 Aug 2026 12:13:18 +0000 Subject: [PATCH] feat(core): Refuse durable pollers when an active workflow has duplicate trigger node ids (#36340) Co-authored-by: Claude Fable 5 --- .../repositories/poller-state.repository.ts | 16 +- .../@n8n/telemetry/src/events/instance.ts | 13 + .../cli/src/commands/__tests__/start.test.ts | 9 + packages/cli/src/commands/start.ts | 7 +- .../__tests__/poll-job-provider.test.ts | 32 ++- .../poll-trigger-task-handler.test.ts | 28 ++ .../poll-trigger-node/poll-job-provider.ts | 8 +- .../poll-trigger-task-handler.ts | 16 ++ .../durable-poller-gate.service.test.ts | 260 ++++++++++++++++++ .../__tests__/poll-cursor.service.test.ts | 26 +- .../triggers/__tests__/trigger-test-utils.ts | 5 + .../triggers/durable-poller-gate.service.ts | 102 +++++++ .../workflows/triggers/poll-cursor.service.ts | 5 +- .../executions/poll-cursor-atomicity.test.ts | 7 +- .../workflows/durable-poller-gate.test.ts | 170 ++++++++++++ 15 files changed, 691 insertions(+), 13 deletions(-) create mode 100644 packages/cli/src/workflows/triggers/__tests__/durable-poller-gate.service.test.ts create mode 100644 packages/cli/src/workflows/triggers/durable-poller-gate.service.ts create mode 100644 packages/cli/test/integration/workflows/durable-poller-gate.test.ts diff --git a/packages/@n8n/db/src/repositories/poller-state.repository.ts b/packages/@n8n/db/src/repositories/poller-state.repository.ts index dd774dabfe0..96d0581869c 100644 --- a/packages/@n8n/db/src/repositories/poller-state.repository.ts +++ b/packages/@n8n/db/src/repositories/poller-state.repository.ts @@ -1,6 +1,6 @@ import { ScheduledTaskStatus } from '@n8n/constants'; import { Service } from '@n8n/di'; -import { DataSource, type EntityManager, type ObjectLiteral } from '@n8n/typeorm'; +import { DataSource, In, type EntityManager, type ObjectLiteral } from '@n8n/typeorm'; import type { QueryDeepPartialEntity } from '@n8n/typeorm/query-builder/QueryPartialEntity'; import { UnexpectedError } from 'n8n-workflow'; @@ -120,6 +120,20 @@ export class PollerStateRepository extends BaseRepository { }); } + /** + * Removes all stored cursors of the given workflows and returns how many + * rows were deleted. Used when durable pollers are refused for the instance: + * with the gate closed, a node whose row is gone falls back to its + * static-data cursor for good. + */ + async deleteWorkflowCursors(workflowIds: string[], ctx: OperationContext = {}): Promise { + if (workflowIds.length === 0) return 0; + const result = await this.managerFor(ctx).delete(PollerState, { + workflowId: In(workflowIds), + }); + return result.affected ?? 0; + } + private buildFenceClause( manager: EntityManager, fence: PollLeaseFence, diff --git a/packages/@n8n/telemetry/src/events/instance.ts b/packages/@n8n/telemetry/src/events/instance.ts index 93346b4997a..379da5259f5 100644 --- a/packages/@n8n/telemetry/src/events/instance.ts +++ b/packages/@n8n/telemetry/src/events/instance.ts @@ -97,4 +97,17 @@ export const INSTANCE_TELEMETRY = defineTelemetryEvents({ }), }), }, + INSTANCE_REFUSED_DURABLE_POLLERS: { + name: 'Instance refused durable pollers', + description: + 'Boot scan found active workflows whose published version has duplicate or missing trigger node ids. Durable poll cursors and durable-scheduler poll triggers were disabled instance-wide and the offending workflows poller_state rows were deleted.', + properties: z.object({ + workflow_ids: z + .array(z.string()) + .describe('Active workflows whose published version failed the trigger node id check'), + deleted_cursor_rows: z + .number() + .describe('How many poller_state rows of the offending workflows were deleted'), + }), + }, }); diff --git a/packages/cli/src/commands/__tests__/start.test.ts b/packages/cli/src/commands/__tests__/start.test.ts index c1ec7e4c1f7..b866fe50237 100644 --- a/packages/cli/src/commands/__tests__/start.test.ts +++ b/packages/cli/src/commands/__tests__/start.test.ts @@ -30,6 +30,7 @@ import { PollJobProvider } from '@/scheduling/poll-trigger-node/poll-job-provide import { JwtService } from '@/services/jwt.service'; import { ShutdownService } from '@/shutdown/shutdown.service'; import { TaskRunnerModule } from '@/task-runners/task-runner-module'; +import { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; const authRolesService = mockInstance(AuthRolesService); authRolesService.init.mockResolvedValue(undefined); @@ -73,6 +74,7 @@ communityPackagesService.init.mockResolvedValue(undefined); const taskRunnerModule = mockInstance(TaskRunnerModule); taskRunnerModule.start.mockResolvedValue(undefined); const pollJobProvider = mockInstance(PollJobProvider); +const durablePollerGate = mockInstance(DurablePollerGateService); const instanceSettings = Container.get(InstanceSettings); @@ -135,6 +137,7 @@ describe('Start - AuthRolesService initialization', () => { mockInstance(BinaryDataConfig, { initialize: vi.fn().mockResolvedValue(undefined) }), ); Container.set(PollJobProvider, pollJobProvider); + Container.set(DurablePollerGateService, durablePollerGate); start = new Start(); // @ts-expect-error - Accessing protected property for testing @@ -188,6 +191,12 @@ describe('Start - AuthRolesService initialization', () => { expect(authRolesService.init).toHaveBeenCalledTimes(1); expect(pollJobProvider.init).toHaveBeenCalledTimes(1); + // The gate's verdict must exist before the provider reads it to pick + // the PollJobManager binding. + expect(durablePollerGate.init).toHaveBeenCalledTimes(1); + expect(durablePollerGate.init.mock.invocationCallOrder[0]).toBeLessThan( + pollJobProvider.init.mock.invocationCallOrder[0], + ); }); it('should initialize AuthRolesService when instanceType is main, multi-main enabled, and is leader', async () => { diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index 8da65b9f20e..024ee3e559a 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -10,11 +10,11 @@ import { import { Command } from '@n8n/decorators'; import { Container } from '@n8n/di'; import { McpServer } from '@n8n/n8n-nodes-langchain/mcp/core'; +import { sleep } from '@n8n/utils/sleep'; import glob from 'fast-glob'; import { createReadStream, createWriteStream, existsSync } from 'fs'; import { mkdir } from 'fs/promises'; import { BinaryDataConfig } from 'n8n-core'; -import { sleep } from '@n8n/utils/sleep'; import { jsonParse } from 'n8n-workflow'; import path from 'path'; import replaceStream from 'replacestream'; @@ -42,9 +42,10 @@ import { Server } from '@/server'; import { JwtService } from '@/services/jwt.service'; import { ExecutionsPruningService } from '@/services/pruning/executions-pruning.service'; import { WorkflowHistoryCompactionService } from '@/services/pruning/workflow-history-compaction.service'; -import { WorkflowStatisticsRollupService } from '@/services/workflow-statistics-rollup.service'; import { UrlService } from '@/services/url.service'; +import { WorkflowStatisticsRollupService } from '@/services/workflow-statistics-rollup.service'; import { WaitTracker } from '@/wait-tracker'; +import { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; import { BaseCommand } from './base-command'; @@ -215,6 +216,8 @@ export class Start extends BaseCommand> { Container.get(DeprecationService).warn(); + // Must complete before PollJobProvider.init() reads the verdict below. + await Container.get(DurablePollerGateService).init(); // Resolved lazily at activation time, so this only needs to run before the // first workflow activation. Container.get(PollJobProvider).init(); diff --git a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-job-provider.test.ts b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-job-provider.test.ts index 3c1b4462484..d1d8f9b812e 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-job-provider.test.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-job-provider.test.ts @@ -6,6 +6,8 @@ import { Container } from '@n8n/di'; import { NoOpPollJobManager, PollJobManager } from 'n8n-core'; import { mock } from 'vitest-mock-extended'; +import type { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; + import { PollJobProvider } from '../poll-job-provider'; import type { PollTriggerJobRegistrar } from '../poll-trigger-job-registrar'; @@ -20,6 +22,7 @@ describe('PollJobProvider', () => { schedulerEnabled = true, publicationEnabled = true, enabledForPollTriggers = true, + durablePollersAllowed = true, logger = mockLogger(), } = {}) => new PollJobProvider( @@ -27,6 +30,7 @@ describe('PollJobProvider', () => { mock({ scheduler: { enabled: schedulerEnabled, enabledForPollTriggers } }), mock({ useWorkflowPublicationService: publicationEnabled }), pollTriggerJobRegistrar, + mock({ allowed: durablePollersAllowed }), ); describe('init', () => { @@ -35,30 +39,52 @@ describe('PollJobProvider', () => { schedulerEnabled: true, publicationEnabled: true, enabledForPollTriggers: true, + durablePollersAllowed: true, active: true, }, { schedulerEnabled: false, publicationEnabled: true, enabledForPollTriggers: true, + durablePollersAllowed: true, active: false, }, { schedulerEnabled: true, publicationEnabled: false, enabledForPollTriggers: true, + durablePollersAllowed: true, active: false, }, { schedulerEnabled: true, publicationEnabled: true, enabledForPollTriggers: false, + durablePollersAllowed: true, + active: false, + }, + { + schedulerEnabled: true, + publicationEnabled: true, + enabledForPollTriggers: true, + durablePollersAllowed: false, active: false, }, ] as const)( - 'binds $active for scheduler=$schedulerEnabled publication=$publicationEnabled pollTriggers=$enabledForPollTriggers', - ({ schedulerEnabled, publicationEnabled, enabledForPollTriggers, active }) => { - makeProvider({ schedulerEnabled, publicationEnabled, enabledForPollTriggers }).init(); + 'binds $active for scheduler=$schedulerEnabled publication=$publicationEnabled pollTriggers=$enabledForPollTriggers gate=$durablePollersAllowed', + ({ + schedulerEnabled, + publicationEnabled, + enabledForPollTriggers, + durablePollersAllowed, + active, + }) => { + makeProvider({ + schedulerEnabled, + publicationEnabled, + enabledForPollTriggers, + durablePollersAllowed, + }).init(); const bound = Container.get(PollJobManager); if (active) { diff --git a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts index 7db35ec041d..adb118fa0f3 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts @@ -10,6 +10,7 @@ import type { Mock, MockInstance } from 'vitest'; import { mock } from 'vitest-mock-extended'; import { createNodeTypes } from '@/workflows/triggers/__tests__/trigger-test-utils'; +import type { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; import type { TriggerExecutionContextFactory } from '@/workflows/triggers/trigger-execution-context.factory'; import { isPollTriggerTaskPayload, POLL_TRIGGER_TASK_TYPE } from '../poll-trigger-task'; @@ -25,12 +26,20 @@ describe('PollTriggerTaskHandler', () => { const scopedLogger = mock(); const rootLogger = mock({ scoped: vi.fn().mockReturnValue(scopedLogger) }); + let durablePollersAllowed = true; + const durablePollerGate = { + get allowed() { + return durablePollersAllowed; + }, + } as DurablePollerGateService; + const handler = new PollTriggerTaskHandler( rootLogger, triggerExecutionContextFactory, triggersAndPollers, workflowRepository, errorReporter, + durablePollerGate, ); const onDispatch = vi.fn(); @@ -130,6 +139,25 @@ describe('PollTriggerTaskHandler', () => { .mockResolvedValue(undefined); }); + describe('durable-poller gate', () => { + // A `scheduled_job` row persisted by an earlier boot keeps firing after a + // later boot closed the gate. Activation has fallen back to in-memory + // polling by then, so running the task would poll the same node twice per + // interval. It must be skipped, not thrown: a throw retries to the max + // attempt count and dead-letters every occurrence. + test('skips the occurrence without polling when the gate is closed', async () => { + durablePollersAllowed = false; + + await handler.execute(buildTask(), report); + + expect(triggerExecutionContextFactory.loadPublishedWorkflowData).not.toHaveBeenCalled(); + expect(triggersAndPollers.runPollFunction).not.toHaveBeenCalled(); + expect(onDispatch).not.toHaveBeenCalled(); + + durablePollersAllowed = true; + }); + }); + describe('task type', () => { test('declares the poll-trigger task type it is bound under', () => { expect(handler.taskType).toBe(POLL_TRIGGER_TASK_TYPE); diff --git a/packages/cli/src/scheduling/poll-trigger-node/poll-job-provider.ts b/packages/cli/src/scheduling/poll-trigger-node/poll-job-provider.ts index 23f1ad3bee8..35fd2195808 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/poll-job-provider.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/poll-job-provider.ts @@ -3,6 +3,8 @@ import { GlobalConfig, WorkflowsConfig } from '@n8n/config'; import { Container, Service } from '@n8n/di'; import { NoOpPollJobManager, PollJobManager } from 'n8n-core'; +import { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; + import { PollTriggerJobRegistrar } from './poll-trigger-job-registrar'; /** @@ -18,6 +20,7 @@ export class PollJobProvider { private readonly globalConfig: GlobalConfig, private readonly workflowsConfig: WorkflowsConfig, private readonly pollTriggerJobRegistrar: PollTriggerJobRegistrar, + private readonly durablePollerGateService: DurablePollerGateService, ) { this.logger = this.logger.scoped('scheduler'); } @@ -26,7 +29,10 @@ export class PollJobProvider { init(): void { const intercepting = this.globalConfig.scheduler.enabled && this.workflowsConfig.useWorkflowPublicationService; - const active = intercepting && this.globalConfig.scheduler.enabledForPollTriggers; + const active = + intercepting && + this.globalConfig.scheduler.enabledForPollTriggers && + this.durablePollerGateService.allowed; if ( this.globalConfig.scheduler.enabled && diff --git a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts index 1f47ce76b75..e273e549d4b 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts @@ -12,6 +12,7 @@ import { import type { INode, IWorkflowBase } from 'n8n-workflow'; import { UnexpectedError } from 'n8n-workflow'; +import { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; import { TriggerExecutionContextFactory } from '@/workflows/triggers/trigger-execution-context.factory'; import { @@ -40,6 +41,7 @@ export class PollTriggerTaskHandler implements TaskHandler { private readonly triggersAndPollers: TriggersAndPollers, private readonly workflowRepository: WorkflowRepository, private readonly errorReporter: ErrorReporter, + private readonly durablePollerGate: DurablePollerGateService, ) { this.logger = this.logger.scoped('scheduler'); } @@ -48,6 +50,20 @@ export class PollTriggerTaskHandler implements TaskHandler { // A setup failure here retries to N8N_SCHEDULER_MAX_ATTEMPTS then dead-letters, // unlike a `poll()` runtime failure below, which routes to the error workflow instead. const { workflowId, nodeId } = this.parsePayload(task); + + // Job rows persisted before the gate closed keep being materialized, and + // activation has fallen back to in-memory polling; running the task too + // would poll the node twice per interval. Skipped, not thrown — a throw + // would retry to the max attempt count and dead-letter every occurrence. + if (!this.durablePollerGate.allowed) { + this.logger.debug('Durable pollers are refused on this instance; skipping the occurrence', { + taskId: task.id, + jobId: task.jobId, + workflowId, + nodeId, + }); + return report.notDispatched(); + } // bypassCache: the poll cursor in staticData must be read live, not from the publish-time cache. const workflowData = await this.triggerExecutionContextFactory.loadPublishedWorkflowData( workflowId, diff --git a/packages/cli/src/workflows/triggers/__tests__/durable-poller-gate.service.test.ts b/packages/cli/src/workflows/triggers/__tests__/durable-poller-gate.service.test.ts new file mode 100644 index 00000000000..3dc1b649753 --- /dev/null +++ b/packages/cli/src/workflows/triggers/__tests__/durable-poller-gate.service.test.ts @@ -0,0 +1,260 @@ +import type { + PollerStateRepository, + WorkflowEntity, + WorkflowHistory, + WorkflowRepository, +} from '@n8n/db'; +import { TELEMETRY_EVENT } from '@n8n/telemetry'; +import type { INode } from 'n8n-workflow'; +import { mock } from 'vitest-mock-extended'; + +import type { Telemetry } from '@/telemetry'; +import { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; +import type { + PublishedWorkflowData, + WorkflowPublishedDataService, +} from '@/workflows/workflow-published-data.service'; +import { WorkflowValidationService } from '@/workflows/workflow-validation.service'; + +import { createNodeTypes, logger, node } from './trigger-test-utils'; + +describe('DurablePollerGateService', () => { + const workflowRepository = mock(); + const publishedDataService = mock(); + const pollerStateRepository = mock(); + const telemetry = mock(); + // Real validation service: only `validateTriggerNodeIds` is exercised, which + // touches none of the constructor dependencies. + const validationService = new WorkflowValidationService(mock(), mock(), mock(), mock()); + const nodeTypes = createNodeTypes(); + + const buildService = () => + new DurablePollerGateService( + logger, + workflowRepository, + publishedDataService, + validationService, + nodeTypes, + pollerStateRepository, + telemetry, + ); + + const published = (nodes: INode[]): PublishedWorkflowData => ({ + workflow: mock(), + publishedVersion: mock({ nodes, connections: {} }), + }); + + /** Declares the active workflows and their published nodes; `null` = active but no published version. */ + const givenActiveWorkflows = (workflows: Record) => { + workflowRepository.getActiveIds.mockResolvedValue(Object.keys(workflows)); + publishedDataService.getPublishedWorkflowData.mockImplementation(async (workflowId) => { + const nodes = workflows[workflowId]; + return await Promise.resolve(nodes ? published(nodes) : null); + }); + }; + + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('blocks durable pollers until the first scan has run', () => { + expect(buildService().allowed).toBe(false); + }); + + describe('init', () => { + it('allows durable pollers when no workflow is active', async () => { + givenActiveWorkflows({}); + const service = buildService(); + + await service.init(); + + expect(service.allowed).toBe(true); + }); + + it('allows durable pollers when all active workflows have unique trigger node ids', async () => { + givenActiveWorkflows({ + 'wf-1': [node('trigger-1', 'poll'), node('trigger-2', 'trigger')], + // Cursor state is keyed on (workflowId, nodeId), so the same node id + // in a *different* workflow is not a collision. + 'wf-2': [node('trigger-1', 'poll')], + }); + const service = buildService(); + + await service.init(); + + expect(service.allowed).toBe(true); + }); + + it('checks the published version of every active workflow', async () => { + givenActiveWorkflows({ + 'wf-1': [node('trigger-1', 'poll')], + 'wf-2': [node('trigger-2', 'poll')], + }); + const service = buildService(); + + await service.init(); + + expect(publishedDataService.getPublishedWorkflowData).toHaveBeenCalledWith('wf-1'); + expect(publishedDataService.getPublishedWorkflowData).toHaveBeenCalledWith('wf-2'); + }); + + it('refuses durable pollers when an active workflow has two triggers sharing a node id', async () => { + givenActiveWorkflows({ + 'wf-clean': [node('trigger-1', 'poll')], + 'wf-dup': [ + node('dup-id', 'poll', { name: 'Poll A' }), + node('dup-id', 'poll', { name: 'Poll B' }), + ], + }); + const service = buildService(); + + await service.init(); + + expect(service.allowed).toBe(false); + }); + + it('refuses durable pollers when an active workflow has a trigger without a node id', async () => { + givenActiveWorkflows({ + 'wf-1': [node('', 'poll', { name: 'Poll without id' })], + }); + const service = buildService(); + + await service.init(); + + expect(service.allowed).toBe(false); + }); + + it('ignores duplicate ids on non-trigger nodes', async () => { + givenActiveWorkflows({ + 'wf-1': [ + node('dup-id', 'noOp', { name: 'Set A' }), + node('dup-id', 'noOp', { name: 'Set B' }), + node('trigger-1', 'poll'), + ], + }); + const service = buildService(); + + await service.init(); + + expect(service.allowed).toBe(true); + }); + + it('skips active workflows that have no published version', async () => { + givenActiveWorkflows({ 'wf-1': null }); + const service = buildService(); + + await service.init(); + + expect(service.allowed).toBe(true); + }); + + // Cursor usage is sticky: `resolveCursor` prefers an existing `poller_state` + // row even when durable cursors are off, so closing the gate alone would not + // stop two duplicate-id nodes from sharing one row. Deleting the offenders' + // rows makes the fallback to static-data cursors terminal. + it('deletes the poller_state rows of offending workflows only', async () => { + givenActiveWorkflows({ + 'wf-clean': [node('trigger-1', 'poll')], + 'wf-dup': [ + node('dup-id', 'poll', { name: 'Poll A' }), + node('dup-id', 'poll', { name: 'Poll B' }), + ], + 'wf-no-id': [node('', 'poll', { name: 'Poll without id' })], + }); + const service = buildService(); + + await service.init(); + + expect(pollerStateRepository.deleteWorkflowCursors).toHaveBeenCalledTimes(1); + expect(pollerStateRepository.deleteWorkflowCursors).toHaveBeenCalledWith([ + 'wf-dup', + 'wf-no-id', + ]); + }); + + it('does not touch poller_state, log, or telemetry when every active workflow is clean', async () => { + givenActiveWorkflows({ 'wf-1': [node('trigger-1', 'poll')] }); + const service = buildService(); + + await service.init(); + + expect(pollerStateRepository.deleteWorkflowCursors).not.toHaveBeenCalled(); + expect(logger.error).not.toHaveBeenCalled(); + expect(telemetry.track).not.toHaveBeenCalled(); + }); + + // "Log loudly, naming the offending workflows" — the log line is the + // operator's only pointer to what to fix, so the ids must be in the message + // itself, not just in metadata. + it('logs an error naming the offending workflows', async () => { + givenActiveWorkflows({ + 'wf-dup': [ + node('dup-id', 'poll', { name: 'Poll A' }), + node('dup-id', 'poll', { name: 'Poll B' }), + ], + 'wf-no-id': [node('', 'poll', { name: 'Poll without id' })], + }); + const service = buildService(); + + await service.init(); + + expect(logger.error).toHaveBeenCalledTimes(1); + const [message] = logger.error.mock.calls[0]; + expect(message).toContain('wf-dup'); + expect(message).toContain('wf-no-id'); + }); + + // A workflow that cannot be scanned (e.g. an uninstalled community node makes + // node-type resolution throw) must never crash startup. It cannot be verified + // either, so the gate stays closed — but its rows are kept: without a scan + // there is no confirmed duplicate to justify deleting cursor state. + it('refuses durable pollers without crashing or deleting rows when a workflow cannot be scanned', async () => { + givenActiveWorkflows({ + 'wf-clean': [node('trigger-1', 'poll')], + 'wf-broken': [node('node-1', 'unrecognized'), node('trigger-2', 'poll')], + }); + const service = buildService(); + + await expect(service.init()).resolves.not.toThrow(); + + expect(service.allowed).toBe(false); + expect(pollerStateRepository.deleteWorkflowCursors).not.toHaveBeenCalled(); + expect(logger.error).toHaveBeenCalledTimes(1); + expect(logger.error.mock.calls[0][0]).toContain('wf-broken'); + }); + + it('still deletes confirmed offenders when another workflow cannot be scanned', async () => { + givenActiveWorkflows({ + 'wf-broken': [node('node-1', 'unrecognized')], + 'wf-dup': [ + node('dup-id', 'poll', { name: 'Poll A' }), + node('dup-id', 'poll', { name: 'Poll B' }), + ], + }); + const service = buildService(); + + await service.init(); + + expect(service.allowed).toBe(false); + expect(pollerStateRepository.deleteWorkflowCursors).toHaveBeenCalledWith(['wf-dup']); + }); + + it('reports the refusal to telemetry with the offending workflow ids and deleted row count', async () => { + givenActiveWorkflows({ + 'wf-dup': [ + node('dup-id', 'poll', { name: 'Poll A' }), + node('dup-id', 'poll', { name: 'Poll B' }), + ], + }); + pollerStateRepository.deleteWorkflowCursors.mockResolvedValue(1); + const service = buildService(); + + await service.init(); + + expect(telemetry.track).toHaveBeenCalledWith( + TELEMETRY_EVENT.INSTANCE.INSTANCE_REFUSED_DURABLE_POLLERS, + { workflow_ids: ['wf-dup'], deleted_cursor_rows: 1 }, + ); + }); + }); +}); diff --git a/packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts b/packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts index e46b9101d90..f4dc7720c81 100644 --- a/packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts @@ -11,6 +11,7 @@ import { mock, type MockProxy } from 'vitest-mock-extended'; import { DuplicateExecutionError } from '@/errors/duplicate-execution.error'; import type { ExecutionPersistence } from '@/executions/execution-persistence'; +import type { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; import { PollCursorService } from '@/workflows/triggers/poll-cursor.service'; describe('PollCursorService', () => { @@ -19,7 +20,7 @@ describe('PollCursorService', () => { let txRunner: MockProxy; - const buildService = (durableCursorsEnabled = true) => { + const buildService = (durableCursorsEnabled = true, durablePollersAllowed = true) => { txRunner = mock(); txRunner.run.mockImplementation( async (ctx: OperationContext, fn: (ctx: OperationContext) => Promise) => await fn(ctx), @@ -30,6 +31,7 @@ describe('PollCursorService', () => { txRunner, executionPersistence, mock({ durableCursorsEnabled }), + mock({ allowed: durablePollersAllowed }), ); }; @@ -47,9 +49,11 @@ describe('PollCursorService', () => { }); describe('enabled', () => { - it('reports the configured flag', () => { - expect(buildService(true).enabled).toBe(true); - expect(buildService(false).enabled).toBe(false); + it('requires both the config flag and the duplicate-id gate', () => { + expect(buildService(true, true).enabled).toBe(true); + expect(buildService(true, false).enabled).toBe(false); + expect(buildService(false, true).enabled).toBe(false); + expect(buildService(false, false).enabled).toBe(false); }); }); @@ -88,6 +92,20 @@ describe('PollCursorService', () => { expect(pollerStateRepository.getOrCreateCursor).not.toHaveBeenCalled(); }); + // Pins the ticket's sticky-row remedy: with the gate refusing, a deleted + // `poller_state` row must never be recreated, even though the flag is on — + // otherwise deleting an offender's rows would not be terminal. + it('does not create a row when the flag is on but the gate refuses durable pollers', async () => { + const service = buildService(true, false); + pollerStateRepository.findCursor.mockResolvedValue(null); + + await expect(service.resolveCursor('wf-1', 'node-1', {})).resolves.toEqual({ + migrated: false, + }); + + expect(pollerStateRepository.getOrCreateCursor).not.toHaveBeenCalled(); + }); + it('still prefers an existing row when the flag is off', async () => { const service = buildService(false); pollerStateRepository.findCursor.mockResolvedValue({ lastItemId: 'from-db' }); diff --git a/packages/cli/src/workflows/triggers/__tests__/trigger-test-utils.ts b/packages/cli/src/workflows/triggers/__tests__/trigger-test-utils.ts index 923a039f242..0cb8d78d49e 100644 --- a/packages/cli/src/workflows/triggers/__tests__/trigger-test-utils.ts +++ b/packages/cli/src/workflows/triggers/__tests__/trigger-test-utils.ts @@ -25,6 +25,11 @@ export function node(id: string, type: string, overrides: Partial = {}): export function createNodeTypes() { const nodeTypes = mock(); nodeTypes.getByNameAndVersion.mockImplementation((type: string) => { + // Mirrors the real NodeTypes, which throws for a node type that is not + // installed on this instance (e.g. an uninstalled community node). + if (type === 'unrecognized') { + throw new Error(`Unrecognized node type: ${type}`); + } if (type === 'trigger') { return { description: { ...description, name: 'trigger' }, trigger: vi.fn() } as never; } diff --git a/packages/cli/src/workflows/triggers/durable-poller-gate.service.ts b/packages/cli/src/workflows/triggers/durable-poller-gate.service.ts new file mode 100644 index 00000000000..1f5595a8fbe --- /dev/null +++ b/packages/cli/src/workflows/triggers/durable-poller-gate.service.ts @@ -0,0 +1,102 @@ +import { Logger } from '@n8n/backend-common'; +import { PollerStateRepository, WorkflowRepository } from '@n8n/db'; +import { Service } from '@n8n/di'; +import { TELEMETRY_EVENT } from '@n8n/telemetry'; +import { ensureError } from '@n8n/utils/errors/ensure-error'; + +import { NodeTypes } from '@/node-types'; +import { Telemetry } from '@/telemetry'; + +import { WorkflowPublishedDataService } from '../workflow-published-data.service'; +import { WorkflowValidationService } from '../workflow-validation.service'; +import { getEnabledTriggerNodes } from './enabled-trigger-nodes'; + +/** + * Boot-time safety check for durable pollers: refuses both durable cursors and + * durable-scheduler poll triggers instance-wide when an active workflow's + * published version has duplicate or missing trigger node ids — two such nodes + * would share one `poller_state` row and one durable job. Deliberately + * disposable: becomes dead code when CAT-4056's heal-before-activate lands. + */ +@Service() +export class DurablePollerGateService { + /** Fail-closed: durable pollers stay off until a scan has completed clean. */ + private allowed_ = false; + + get allowed() { + return this.allowed_; + } + + constructor( + private readonly logger: Logger, + private readonly workflowRepository: WorkflowRepository, + private readonly workflowPublishedDataService: WorkflowPublishedDataService, + private readonly workflowValidationService: WorkflowValidationService, + private readonly nodeTypes: NodeTypes, + private readonly pollerStateRepository: PollerStateRepository, + private readonly telemetry: Telemetry, + ) { + this.logger = this.logger.scoped('poll-trigger'); + } + + /** + * Boot-only by design: mid-process arrivals of faulty data (source-control + * pull, legacy-path republish) are an accepted residual until CAT-4056's + * heal-before-activate checks at import/activation time. A leader-takeover + * re-check was deliberately dropped — it would remediate at an arbitrary + * point after the harm began. + */ + async init() { + const ids = await this.workflowRepository.getActiveIds(); + const offenders: Array<{ workflowId: string; detail: string }> = []; + const unscannable: Array<{ workflowId: string; detail: string }> = []; + + for (const id of ids) { + try { + const data = await this.workflowPublishedDataService.getPublishedWorkflowData(id); + if (data === null) { + continue; + } + + const triggerNodes = getEnabledTriggerNodes(data.publishedVersion, this.nodeTypes); + + const result = this.workflowValidationService.validateTriggerNodeIds(triggerNodes); + + if (!result.isValid) { + offenders.push({ workflowId: id, detail: result.error ?? 'invalid trigger node ids' }); + } + } catch (error) { + // A workflow that cannot be scanned (e.g. node-type resolution throws + // for an uninstalled community node) must never crash startup. It + // cannot be verified either, so it closes the gate — but its rows are + // kept: without a scan there is no confirmed duplicate to justify + // deleting cursor state. + unscannable.push({ workflowId: id, detail: ensureError(error).message }); + } + } + + this.allowed_ = offenders.length === 0 && unscannable.length === 0; + + if (unscannable.length > 0) { + this.logger.error( + `Durable pollers are disabled on this instance: active workflows [${unscannable.map(({ workflowId }) => workflowId).join(', ')}] could not be scanned for duplicate trigger node ids. Fix the reported error, e.g. by installing the missing node package, then restart.`, + { unscannable }, + ); + } + + if (offenders.length > 0) { + const workflowIds = offenders.map(({ workflowId }) => workflowId); + // Deletion is terminal only because the gate keeps `enabled` false — + // otherwise the next poll would recreate the rows via getOrCreateCursor. + const deletedCursorRows = await this.pollerStateRepository.deleteWorkflowCursors(workflowIds); + this.logger.error( + `Durable pollers are disabled on this instance: active workflows [${workflowIds.join(', ')}] have duplicate or missing trigger node ids in their published version. Re-publish them after removing and re-adding the affected trigger nodes.`, + { offenders, deletedCursorRows }, + ); + this.telemetry.track(TELEMETRY_EVENT.INSTANCE.INSTANCE_REFUSED_DURABLE_POLLERS, { + workflow_ids: workflowIds, + deleted_cursor_rows: deletedCursorRows, + }); + } + } +} diff --git a/packages/cli/src/workflows/triggers/poll-cursor.service.ts b/packages/cli/src/workflows/triggers/poll-cursor.service.ts index 039f3d22999..a8b015ea447 100644 --- a/packages/cli/src/workflows/triggers/poll-cursor.service.ts +++ b/packages/cli/src/workflows/triggers/poll-cursor.service.ts @@ -6,6 +6,8 @@ import type { PollCursor } from 'n8n-workflow'; import { ExecutionPersistence } from '@/executions/execution-persistence'; +import { DurablePollerGateService } from './durable-poller-gate.service'; + /** Narrows a stored cursor, which the persistence layer types more loosely. */ const toPollCursor = (cursor: PollerCursor): PollCursor => cursor as PollCursor; @@ -16,10 +18,11 @@ export class PollCursorService { private readonly transactionRunner: TransactionRunner, private readonly executionPersistence: ExecutionPersistence, private readonly pollerConfig: PollerConfig, + private readonly durablePollerGateService: DurablePollerGateService, ) {} get enabled(): boolean { - return this.pollerConfig.durableCursorsEnabled; + return this.pollerConfig.durableCursorsEnabled && this.durablePollerGateService.allowed; } /** diff --git a/packages/cli/test/integration/executions/poll-cursor-atomicity.test.ts b/packages/cli/test/integration/executions/poll-cursor-atomicity.test.ts index 45567a6b33e..ce46ec21a14 100644 --- a/packages/cli/test/integration/executions/poll-cursor-atomicity.test.ts +++ b/packages/cli/test/integration/executions/poll-cursor-atomicity.test.ts @@ -1,4 +1,4 @@ -import { createWorkflow, testDb } from '@n8n/backend-test-utils'; +import { createWorkflow, mockInstance, testDb } from '@n8n/backend-test-utils'; import { PollerConfig } from '@n8n/config'; import type { CreateExecutionPayload, @@ -20,10 +20,15 @@ import { createEmptyRunExecutionData } from 'n8n-workflow'; import { ExecutionPersistence } from '@/executions/execution-persistence'; import { POLL_TRIGGER_TASK_TYPE } from '@/scheduling/poll-trigger-node/poll-trigger-task'; +import { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; import { PollCursorService } from '@/workflows/triggers/poll-cursor.service'; import { createDueJobFactory, seedDueTask } from '../scheduling/shared/job-factory'; +// The duplicate-trigger-id gate is fail-closed until a boot scan runs; open it +// so the config flag alone controls the paths under test. +mockInstance(DurablePollerGateService, { allowed: true }); + describe('poll cursor atomicity', () => { const nodeId = 'node-1'; diff --git a/packages/cli/test/integration/workflows/durable-poller-gate.test.ts b/packages/cli/test/integration/workflows/durable-poller-gate.test.ts new file mode 100644 index 00000000000..3b0b49ab61f --- /dev/null +++ b/packages/cli/test/integration/workflows/durable-poller-gate.test.ts @@ -0,0 +1,170 @@ +import { Logger } from '@n8n/backend-common'; +import { + createWorkflowWithHistory, + mockInstance, + setActiveVersion, + testDb, +} from '@n8n/backend-test-utils'; +import { + PollerStateRepository, + WorkflowPublishedVersionRepository, + WorkflowRepository, +} from '@n8n/db'; +import { TELEMETRY_EVENT } from '@n8n/telemetry'; +import { Container } from '@n8n/di'; +import type { INode } from 'n8n-workflow'; +import { v4 as uuid } from 'uuid'; + +import { NodeTypes } from '@/node-types'; +import { Telemetry } from '@/telemetry'; +import { DurablePollerGateService } from '@/workflows/triggers/durable-poller-gate.service'; +import { WorkflowPublishedDataService } from '@/workflows/workflow-published-data.service'; +import { WorkflowValidationService } from '@/workflows/workflow-validation.service'; + +import { createOwner } from '../shared/db/users'; +import * as utils from '../shared/utils'; +import { loadNodesFromDist } from '../shared/utils/node-types-data'; + +const telemetry = mockInstance(Telemetry); + +/** + * The gate against a real database: published versions are read from the + * `workflow_published_version` mapping, the verdict flips on real duplicate + * trigger node ids, and the offenders' `poller_state` rows are really deleted + * while clean workflows keep theirs. The unit suite mocks the repositories; + * this proves the seams line up at runtime. + */ +describe('DurablePollerGateService (integration)', () => { + let owner: Awaited>; + let pollerStateRepository: PollerStateRepository; + let publishedVersionRepository: WorkflowPublishedVersionRepository; + + beforeAll(async () => { + await testDb.init(); + await utils.initNodeTypes(loadNodesFromDist(['n8n-nodes-base.scheduleTrigger'])); + owner = await createOwner(); + pollerStateRepository = Container.get(PollerStateRepository); + publishedVersionRepository = Container.get(WorkflowPublishedVersionRepository); + }); + + afterEach(async () => { + // Delete WorkflowPublishedVersion first: it references WorkflowHistory with + // onDelete RESTRICT, and deleting WorkflowEntity cascades into WorkflowHistory. + await testDb.truncate([ + 'WorkflowPublishedVersion', + 'WorkflowPublishHistory', + 'PollerState', + 'WorkflowEntity', + 'WorkflowHistory', + ]); + }); + + afterAll(async () => { + await testDb.terminate(); + }); + + // The gate holds its verdict as instance state, so every test gets a fresh + // instance instead of the container singleton. + const buildGate = () => + new DurablePollerGateService( + Container.get(Logger), + Container.get(WorkflowRepository), + Container.get(WorkflowPublishedDataService), + Container.get(WorkflowValidationService), + Container.get(NodeTypes), + pollerStateRepository, + Container.get(Telemetry), + ); + + const scheduleTrigger = (name: string, id = uuid()): INode => ({ + id, + name, + type: 'n8n-nodes-base.scheduleTrigger', + typeVersion: 1, + position: [0, 0], + parameters: {}, + }); + + // All three parts the gate reads: workflow (active), history version, and + // the published-version mapping pointing at it. + const createPublishedWorkflow = async (nodes: INode[]) => { + const workflow = await createWorkflowWithHistory( + { active: true, nodes, connections: {} }, + owner, + ); + await setActiveVersion(workflow.id, workflow.versionId); + await publishedVersionRepository.setPublishedVersion(workflow.id, workflow.versionId); + return workflow; + }; + + const seedCursor = async (workflowId: string, nodeId: string) => + await pollerStateRepository.insert({ workflowId, nodeId, cursor: { lastItemId: 'seeded' } }); + + test('keeps durable pollers allowed and cursors intact on a clean instance', async () => { + const trigger = scheduleTrigger('Trigger A'); + const workflow = await createPublishedWorkflow([trigger, scheduleTrigger('Trigger B')]); + await seedCursor(workflow.id, trigger.id); + + const gate = buildGate(); + await gate.init(); + + expect(gate.allowed).toBe(true); + await expect(pollerStateRepository.findCursor(workflow.id, trigger.id)).resolves.toEqual({ + lastItemId: 'seeded', + }); + }); + + // Only n8n-nodes-base.scheduleTrigger is loaded in this suite, so the noOp + // node below makes the real NodeTypes throw UnrecognizedNodeTypeError — the + // uninstalled-community-node case. Startup must survive it. + test('refuses durable pollers without crashing when a workflow has an uninstalled node type', async () => { + await createPublishedWorkflow([ + scheduleTrigger('Trigger A'), + { + id: uuid(), + name: 'NoOp', + type: 'n8n-nodes-base.noOp', + typeVersion: 1, + position: [0, 0], + parameters: {}, + }, + ]); + + const gate = buildGate(); + await expect(gate.init()).resolves.not.toThrow(); + + expect(gate.allowed).toBe(false); + }); + + test('refuses durable pollers and deletes only the offender cursor rows', async () => { + const cleanTrigger = scheduleTrigger('Clean Trigger'); + const cleanWorkflow = await createPublishedWorkflow([cleanTrigger]); + await seedCursor(cleanWorkflow.id, cleanTrigger.id); + + const duplicateId = uuid(); + const offender = await createPublishedWorkflow([ + scheduleTrigger('Poll A', duplicateId), + scheduleTrigger('Poll B', duplicateId), + ]); + // The one row both duplicate-id nodes would contend on. + await seedCursor(offender.id, duplicateId); + + const gate = buildGate(); + await gate.init(); + + expect(gate.allowed).toBe(false); + await expect(pollerStateRepository.findCursor(offender.id, duplicateId)).resolves.toBeNull(); + await expect( + pollerStateRepository.findCursor(cleanWorkflow.id, cleanTrigger.id), + ).resolves.toEqual({ lastItemId: 'seeded' }); + // The deleted-row count reported to telemetry is the real DELETE's count. + expect(telemetry.track).toHaveBeenCalledWith( + TELEMETRY_EVENT.INSTANCE.INSTANCE_REFUSED_DURABLE_POLLERS, + { workflow_ids: [offender.id], deleted_cursor_rows: 1 }, + ); + }); + + test('deleting cursors for no workflows touches nothing and reports zero rows', async () => { + await expect(pollerStateRepository.deleteWorkflowCursors([])).resolves.toBe(0); + }); +});