feat(core): Refuse durable pollers when an active workflow has duplicate trigger node ids (#36340)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

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<PollerState> {
 		});
 	}
 
+	/**
+	 * 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<number> {
+		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,

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'),
+		}),
+	},
 });

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 () => {

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<z.infer<typeof flagsSchema>> {
 
 		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();

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<GlobalConfig>({ scheduler: { enabled: schedulerEnabled, enabledForPollTriggers } }),
 			mock<WorkflowsConfig>({ useWorkflowPublicationService: publicationEnabled }),
 			pollTriggerJobRegistrar,
+			mock<DurablePollerGateService>({ 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) {

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<Logger>();
 	const rootLogger = mock<Logger>({ 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);

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 &&

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,

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<WorkflowRepository>();
+	const publishedDataService = mock<WorkflowPublishedDataService>();
+	const pollerStateRepository = mock<PollerStateRepository>();
+	const telemetry = mock<Telemetry>();
+	// 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<WorkflowEntity>(),
+		publishedVersion: mock<WorkflowHistory>({ nodes, connections: {} }),
+	});
+
+	/** Declares the active workflows and their published nodes; `null` = active but no published version. */
+	const givenActiveWorkflows = (workflows: Record<string, INode[] | null>) => {
+		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 },
+			);
+		});
+	});
+});

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<TransactionRunner>;
 
-	const buildService = (durableCursorsEnabled = true) => {
+	const buildService = (durableCursorsEnabled = true, durablePollersAllowed = true) => {
 		txRunner = mock<TransactionRunner>();
 		txRunner.run.mockImplementation(
 			async <T>(ctx: OperationContext, fn: (ctx: OperationContext) => Promise<T>) => await fn(ctx),
@@ -30,6 +31,7 @@ describe('PollCursorService', () => {
 			txRunner,
 			executionPersistence,
 			mock<PollerConfig>({ durableCursorsEnabled }),
+			mock<DurablePollerGateService>({ 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' });

packages/cli/src/workflows/triggers/__tests__/trigger-test-utils.ts

@@ -25,6 +25,11 @@ export function node(id: string, type: string, overrides: Partial<INode> = {}):
 export function createNodeTypes() {
 	const nodeTypes = mock<NodeTypes>();
 	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;
 		}

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,
+			});
+		}
+	}
+}

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;
 	}
 
 	/**

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';
 

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<ReturnType<typeof createOwner>>;
+	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);
+	});
+});