perf(core): Read the poller state row once per durable poll tick (#36973)

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

packages/@n8n/db/src/repositories/index.ts

@@ -59,6 +59,7 @@ export {
 	type PollerCursor,
 	type PollLeaseFence,
 	type PollerFailureState,
+	type PollerFullState,
 } from './poller-state.repository';
 export { ProcessedDataRepository } from './processed-data.repository';
 export { SettingsRepository } from './settings.repository';

packages/@n8n/db/src/repositories/poller-state.repository.ts

@@ -21,6 +21,10 @@ export interface PollerFailureState {
 	backoffUntil: Date | null;
 }
 
+export interface PollerFullState extends PollerFailureState {
+	cursor: PollerCursor;
+}
+
 @Service()
 export class PollerStateRepository extends BaseRepository<PollerState> {
 	private readonly isPostgres: boolean;
@@ -151,19 +155,23 @@ export class PollerStateRepository extends BaseRepository<PollerState> {
 		return result.affected ?? 0;
 	}
 
-	/** The node's failure counters, or `null` if it has no stored row. */
-	async findFailureState(
+	/** The node's cursor plus failure counters, or `null` if it has no stored row. */
+	async findState(
 		workflowId: string,
 		nodeId: string,
 		ctx: OperationContext = {},
-	): Promise<PollerFailureState | null> {
+	): Promise<PollerFullState | null> {
 		const row = await this.managerFor(ctx).findOne(PollerState, {
-			select: ['consecutiveErrors', 'backoffUntil'],
+			select: ['cursor', 'consecutiveErrors', 'backoffUntil'],
 			where: { workflowId, nodeId },
 		});
 		return row === null
 			? null
-			: { consecutiveErrors: row.consecutiveErrors, backoffUntil: row.backoffUntil };
+			: {
+					cursor: row.cursor,
+					consecutiveErrors: row.consecutiveErrors,
+					backoffUntil: row.backoffUntil,
+				};
 	}
 
 	/**

packages/cli/src/controllers/e2e.controller.ts

@@ -277,12 +277,11 @@ export class E2EController {
 	@Get('/poller-state', { skipAuth: true })
 	async getPollerState(req: Request<{}, {}, {}, { workflowId: string; nodeId: string }>) {
 		const { workflowId, nodeId } = req.query;
-		const cursor = await this.pollerStateRepository.findCursor(workflowId, nodeId);
-		const failureState = await this.pollerStateRepository.findFailureState(workflowId, nodeId);
+		const state = await this.pollerStateRepository.findState(workflowId, nodeId);
 		return {
-			cursor,
-			consecutiveErrors: failureState?.consecutiveErrors ?? 0,
-			backoffUntil: failureState?.backoffUntil ?? null,
+			cursor: state?.cursor ?? null,
+			consecutiveErrors: state?.consecutiveErrors ?? 0,
+			backoffUntil: state?.backoffUntil ?? null,
 		};
 	}
 
@@ -475,7 +474,7 @@ export class E2EController {
 	}
 
 	private static coverageKey(url: string, fn: Profiler.FunctionCoverage): string {
-		return `${url}${fn.functionName}${fn.ranges[0]?.startOffset ?? 0}`;
+		return `${url} ${fn.functionName} ${fn.ranges[0]?.startOffset ?? 0}`;
 	}
 
 	private static coverageCount(fn: Profiler.FunctionCoverage): number {

packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts

@@ -1,7 +1,7 @@
 /* eslint-disable @typescript-eslint/unbound-method */
 import type { Logger } from '@n8n/backend-common';
 import type { GlobalConfig } from '@n8n/config';
-import type { PollerFailureState, WorkflowRepository } from '@n8n/db';
+import type { PollerFullState, WorkflowRepository } from '@n8n/db';
 import { createDispatchReporter, type ClaimedTask } from '@n8n/scheduler';
 import type { ErrorReporter, TriggersAndPollers } from 'n8n-core';
 import type { INode, INodeExecutionData, IPollFunctions, IWorkflowBase } from 'n8n-workflow';
@@ -133,7 +133,7 @@ describe('PollTriggerTaskHandler', () => {
 		triggersAndPollers.runPollFunction.mockResolvedValue(pollData);
 		workflowRepository.isActive.mockResolvedValue(true);
 
-		pollBackoffService.getFailureState.mockResolvedValue(null);
+		pollBackoffService.getState.mockResolvedValue(null);
 		pollBackoffService.isBackingOff.mockReturnValue(false);
 
 		acquireIsolate = vi
@@ -189,6 +189,7 @@ describe('PollTriggerTaskHandler', () => {
 				buildWorkflowData(),
 				triggerNode,
 				{ taskId: 'task-1', leaseEpoch: 1 },
+				undefined,
 			);
 			expect(triggersAndPollers.runPollFunction).toHaveBeenCalledWith(
 				workflow,
@@ -197,6 +198,23 @@ describe('PollTriggerTaskHandler', () => {
 			);
 		});
 
+		test('threads the cursor from the top-of-tick state read into the poll context', async () => {
+			pollBackoffService.getState.mockResolvedValue({
+				cursor: { lastItemId: 'prefetched' },
+				consecutiveErrors: 0,
+				backoffUntil: null,
+			});
+
+			await handler.execute(buildTask(), report);
+
+			expect(triggerExecutionContextFactory.createPollExecutionContext).toHaveBeenCalledWith(
+				buildWorkflowData(),
+				triggerNode,
+				{ taskId: 'task-1', leaseEpoch: 1 },
+				{ lastItemId: 'prefetched' },
+			);
+		});
+
 		test('reads workflow data fresh (non-cached) so the poll cursor is never stale', async () => {
 			await handler.execute(buildTask(), report);
 
@@ -541,11 +559,12 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('skips the tick while backing off, without loading the workflow or polling', async () => {
-			const state: PollerFailureState = {
+			const state: PollerFullState = {
+				cursor: {},
 				consecutiveErrors: 3,
 				backoffUntil: new Date(fixedNow.getTime() + 60_000),
 			};
-			pollBackoffService.getFailureState.mockResolvedValue(state);
+			pollBackoffService.getState.mockResolvedValue(state);
 			pollBackoffService.isBackingOff.mockReturnValue(true);
 
 			const decision = await handler.execute(buildTask(), report);
@@ -558,8 +577,8 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('records a failure and no success when poll() throws', async () => {
-			const state: PollerFailureState = { consecutiveErrors: 1, backoffUntil: null };
-			pollBackoffService.getFailureState.mockResolvedValue(state);
+			const state: PollerFullState = { cursor: {}, consecutiveErrors: 1, backoffUntil: null };
+			pollBackoffService.getState.mockResolvedValue(state);
 			const error = new Error('poll source unreachable');
 			triggersAndPollers.runPollFunction.mockRejectedValue(error);
 
@@ -588,8 +607,8 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('records no failure when the poll returned and a later step throws', async () => {
-			const state: PollerFailureState = { consecutiveErrors: 1, backoffUntil: null };
-			pollBackoffService.getFailureState.mockResolvedValue(state);
+			const state: PollerFullState = { cursor: {}, consecutiveErrors: 1, backoffUntil: null };
+			pollBackoffService.getState.mockResolvedValue(state);
 			const error = new Error('database unavailable');
 			workflowRepository.isActive.mockRejectedValue(error);
 
@@ -602,8 +621,8 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('still clears the failure state when the poll succeeds but committing its cursor fails', async () => {
-			const state: PollerFailureState = { consecutiveErrors: 2, backoffUntil: null };
-			pollBackoffService.getFailureState.mockResolvedValue(state);
+			const state: PollerFullState = { cursor: {}, consecutiveErrors: 2, backoffUntil: null };
+			pollBackoffService.getState.mockResolvedValue(state);
 			triggersAndPollers.runPollFunction.mockResolvedValue(null);
 			pollFunctions.__commitCursor.mockRejectedValue(new Error('poller state write failed'));
 
@@ -620,8 +639,8 @@ describe('PollTriggerTaskHandler', () => {
 			['a poll returning items', pollData],
 			['a poll returning no items', null],
 		])('clears the failure state after %s', async (_name, pollResult) => {
-			const state: PollerFailureState = { consecutiveErrors: 2, backoffUntil: null };
-			pollBackoffService.getFailureState.mockResolvedValue(state);
+			const state: PollerFullState = { cursor: {}, consecutiveErrors: 2, backoffUntil: null };
+			pollBackoffService.getState.mockResolvedValue(state);
 			triggersAndPollers.runPollFunction.mockResolvedValue(pollResult);
 
 			await handler.execute(buildTask(), report);
@@ -635,8 +654,8 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('clears the failure state even when the workflow was deactivated during the poll', async () => {
-			const state: PollerFailureState = { consecutiveErrors: 1, backoffUntil: null };
-			pollBackoffService.getFailureState.mockResolvedValue(state);
+			const state: PollerFullState = { cursor: {}, consecutiveErrors: 1, backoffUntil: null };
+			pollBackoffService.getState.mockResolvedValue(state);
 			workflowRepository.isActive.mockResolvedValue(false);
 
 			await handler.execute(buildTask(), report);
@@ -649,8 +668,8 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('does not record a failure for a workflow deactivated during a failing poll, but still hands off the error', async () => {
-			const state: PollerFailureState = { consecutiveErrors: 1, backoffUntil: null };
-			pollBackoffService.getFailureState.mockResolvedValue(state);
+			const state: PollerFullState = { cursor: {}, consecutiveErrors: 1, backoffUntil: null };
+			pollBackoffService.getState.mockResolvedValue(state);
 			const error = new Error('poll source unreachable');
 			triggersAndPollers.runPollFunction.mockRejectedValue(error);
 			workflowRepository.isActive.mockResolvedValue(false);
@@ -663,8 +682,8 @@ describe('PollTriggerTaskHandler', () => {
 		});
 
 		test('records a failure when the active-state read itself fails, rather than let a real failure go unbacked-off', async () => {
-			const state: PollerFailureState = { consecutiveErrors: 1, backoffUntil: null };
-			pollBackoffService.getFailureState.mockResolvedValue(state);
+			const state: PollerFullState = { cursor: {}, consecutiveErrors: 1, backoffUntil: null };
+			pollBackoffService.getState.mockResolvedValue(state);
 			const error = new Error('poll source unreachable');
 			triggersAndPollers.runPollFunction.mockRejectedValue(error);
 			workflowRepository.isActive.mockRejectedValue(new Error('database unavailable'));
@@ -697,12 +716,12 @@ describe('PollTriggerTaskHandler', () => {
 
 			await expect(handler.execute(task, report)).rejects.toThrow();
 
-			expect(pollBackoffService.getFailureState).not.toHaveBeenCalled();
+			expect(pollBackoffService.getState).not.toHaveBeenCalled();
 		});
 
 		test('still runs the poll when reading the failure state throws', async () => {
 			const error = new Error('poller state read failed');
-			pollBackoffService.getFailureState.mockRejectedValue(error);
+			pollBackoffService.getState.mockRejectedValue(error);
 
 			await handler.execute(buildTask(), report);
 

packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts

@@ -77,9 +77,7 @@ export class PollTriggerTaskHandler implements TaskHandler {
 		const { workflowId, nodeId } = this.parsePayload(task);
 
 		const now = new Date();
-		const state = await this.pollBackoffService
-			.getFailureState(workflowId, nodeId)
-			.catch(() => null);
+		const state = await this.pollBackoffService.getState(workflowId, nodeId).catch(() => null);
 		if (this.pollBackoffService.isBackingOff(state, now)) {
 			this.logger.debug('Poll is backing off; skipping this occurrence', {
 				taskId: task.id,
@@ -113,10 +111,12 @@ export class PollTriggerTaskHandler implements TaskHandler {
 		const node = this.resolveTriggerNode(workflowData, nodeId, task);
 
 		const { workflow, pollFunctions } =
-			await this.triggerExecutionContextFactory.createPollExecutionContext(workflowData, node, {
-				taskId: task.id,
-				leaseEpoch: task.leaseEpoch,
-			});
+			await this.triggerExecutionContextFactory.createPollExecutionContext(
+				workflowData,
+				node,
+				{ taskId: task.id, leaseEpoch: task.leaseEpoch },
+				state?.cursor,
+			);
 
 		// Poll and hand-off share one staging scope, so a cursor staged here can only
 		// be committed by this poll and never by a later occurrence.

packages/cli/src/workflows/triggers/__tests__/poll-backoff.service.test.ts

@@ -1,6 +1,6 @@
 import type { Logger } from '@n8n/backend-common';
 import type { SchedulerConfig, WorkflowsConfig } from '@n8n/config';
-import type { PollerFailureState, PollerStateRepository } from '@n8n/db';
+import type { PollerFailureState, PollerFullState, PollerStateRepository } from '@n8n/db';
 import type { ErrorReporter } from 'n8n-core';
 import { mock } from 'vitest-mock-extended';
 
@@ -57,36 +57,40 @@ describe('PollBackoffService', () => {
 		});
 	});
 
-	describe('getFailureState', () => {
+	describe('getState', () => {
 		test('does not query when the flag is off', async () => {
 			const service = buildService(false);
 
-			await expect(service.getFailureState('wf-1', 'node-1')).resolves.toBeNull();
+			await expect(service.getState('wf-1', 'node-1')).resolves.toBeNull();
 
-			expect(pollerStateRepository.findFailureState).not.toHaveBeenCalled();
+			expect(pollerStateRepository.findState).not.toHaveBeenCalled();
 		});
 
-		test('returns the stored failure state when the flag is on', async () => {
-			const state: PollerFailureState = { consecutiveErrors: 2, backoffUntil: now };
-			pollerStateRepository.findFailureState.mockResolvedValue(state);
+		test('returns the stored state, cursor included, when the flag is on', async () => {
+			const state: PollerFullState = {
+				cursor: { lastItemId: 'a' },
+				consecutiveErrors: 2,
+				backoffUntil: now,
+			};
+			pollerStateRepository.findState.mockResolvedValue(state);
 			const service = buildService();
 
-			await expect(service.getFailureState('wf-1', 'node-1')).resolves.toEqual(state);
+			await expect(service.getState('wf-1', 'node-1')).resolves.toEqual(state);
 		});
 
 		test('returns null for a node with no stored row', async () => {
-			pollerStateRepository.findFailureState.mockResolvedValue(null);
+			pollerStateRepository.findState.mockResolvedValue(null);
 			const service = buildService();
 
-			await expect(service.getFailureState('wf-1', 'node-1')).resolves.toBeNull();
+			await expect(service.getState('wf-1', 'node-1')).resolves.toBeNull();
 		});
 
 		test('swallows a failing read and reports it, returning null instead of throwing', async () => {
 			const readError = new Error('poller state read failed');
-			pollerStateRepository.findFailureState.mockRejectedValue(readError);
+			pollerStateRepository.findState.mockRejectedValue(readError);
 			const service = buildService();
 
-			await expect(service.getFailureState('wf-1', 'node-1')).resolves.toBeNull();
+			await expect(service.getState('wf-1', 'node-1')).resolves.toBeNull();
 
 			expectErrorReported(readError);
 		});

packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts

@@ -140,6 +140,51 @@ describe('PollCursorService', () => {
 			);
 		});
 
+		it('uses a prefetched cursor without any read or transaction when the flag is on', async () => {
+			const service = buildService(true);
+
+			const resolved = await service.resolveCursor(
+				'wf-1',
+				'node-1',
+				{ lastItemId: 'from-static-data' },
+				{ lastItemId: 'prefetched' },
+			);
+
+			expect(resolved).toEqual({ migrated: true, cursor: { lastItemId: 'prefetched' } });
+			expect(pollerStateRepository.getOrCreateCursor).not.toHaveBeenCalled();
+			expect(pollerStateRepository.findCursor).not.toHaveBeenCalled();
+			expect(txRunner.run).not.toHaveBeenCalled();
+		});
+
+		it('treats an empty prefetched cursor as a stored cursor, not a missing one', async () => {
+			const service = buildService(true);
+
+			const resolved = await service.resolveCursor('wf-1', 'node-1', { lastItemId: 'seed' }, {});
+
+			expect(resolved).toEqual({ migrated: true, cursor: {} });
+			expect(pollerStateRepository.getOrCreateCursor).not.toHaveBeenCalled();
+		});
+
+		it('falls back to getOrCreateCursor when no cursor was prefetched', async () => {
+			const service = buildService(true);
+			pollerStateRepository.getOrCreateCursor.mockResolvedValue({ lastItemId: 'from-db' });
+
+			const resolved = await service.resolveCursor(
+				'wf-1',
+				'node-1',
+				{ lastItemId: 'seed' },
+				undefined,
+			);
+
+			expect(resolved).toEqual({ migrated: true, cursor: { lastItemId: 'from-db' } });
+			expect(pollerStateRepository.getOrCreateCursor).toHaveBeenCalledWith(
+				'wf-1',
+				'node-1',
+				{ lastItemId: 'seed' },
+				expect.anything(),
+			);
+		});
+
 		it('does not create a row when the flag is off and the node has never migrated', async () => {
 			const service = buildService(false);
 			pollerStateRepository.findCursor.mockResolvedValue(null);

packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts

@@ -775,13 +775,19 @@ describe('TriggerExecutionContextFactory', () => {
 			return workflow;
 		};
 
-		const buildContext = (workflow: Workflow, node: INode): RunnablePollFunctions => {
+		const buildContext = (
+			workflow: Workflow,
+			node: INode,
+			prefetchedCursor?: Record<string, unknown>,
+		): RunnablePollFunctions => {
 			const getPollFunctions = factory.getExecutePollFunctions(
 				mock<IWorkflowBase>({ id: 'wf-1', name: 'Test Workflow' }),
 				additionalData,
 				mode,
 				activation,
 				async () => mock<IWorkflowBase>({ id: 'wf-1', name: 'Test Workflow' }),
+				undefined,
+				prefetchedCursor,
 			);
 			return getPollFunctions(
 				workflow,
@@ -817,9 +823,12 @@ describe('TriggerExecutionContextFactory', () => {
 				expect(context.getWorkflowStaticData('node')).toEqual({ lastItemId: 'from-db' });
 			});
 
-			expect(pollCursorService.resolveCursor).toHaveBeenCalledWith('wf-1', 'node-1', {
-				lastItemId: 'from-static-data',
-			});
+			expect(pollCursorService.resolveCursor).toHaveBeenCalledWith(
+				'wf-1',
+				'node-1',
+				{ lastItemId: 'from-static-data' },
+				undefined,
+			);
 		});
 
 		test('falls back to the real static data when the node has never migrated and the flag is off', async () => {
@@ -862,9 +871,27 @@ describe('TriggerExecutionContextFactory', () => {
 				});
 			});
 
-			expect(pollCursorService.resolveCursor).toHaveBeenLastCalledWith('wf-1', 'node-1', {
-				lastItemId: 'mutated-before-migration',
-			});
+			expect(pollCursorService.resolveCursor).toHaveBeenLastCalledWith(
+				'wf-1',
+				'node-1',
+				{ lastItemId: 'mutated-before-migration' },
+				undefined,
+			);
+		});
+
+		test('threads a prefetched cursor into resolveCursor so it can skip its own read', async () => {
+			workflow.getStaticData.mockReturnValue({ lastItemId: 'from-static-data' });
+			const prefetched = { lastItemId: 'prefetched' };
+			const prefetchedContext = buildContext(workflow, node, prefetched);
+
+			await prefetchedContext.__runPoll(async () => {});
+
+			expect(pollCursorService.resolveCursor).toHaveBeenCalledWith(
+				'wf-1',
+				'node-1',
+				{ lastItemId: 'from-static-data' },
+				prefetched,
+			);
 		});
 
 		test('routes to runWorkflow, not runPolledWorkflow, when the node leaves its static data unchanged', async () => {
@@ -1314,8 +1341,8 @@ describe('TriggerExecutionContextFactory', () => {
 			});
 
 			// Built with the activation path's execution/activation modes ('trigger'/'update').
-			// No per-occurrence deduplication key is threaded; the trailing argument is the
-			// lease fence, which this path has none of.
+			// No per-occurrence deduplication key is threaded; the trailing arguments are the
+			// lease fence and the prefetched cursor, which this path has neither of.
 			expect(getExecutePollFunctionsSpy).toHaveBeenCalledWith(
 				workflowData,
 				additionalData,
@@ -1323,6 +1350,7 @@ describe('TriggerExecutionContextFactory', () => {
 				'update',
 				expect.any(Function),
 				undefined,
+				undefined,
 			);
 
 			expect(getPollFunctions).toHaveBeenCalledWith(
@@ -1355,6 +1383,33 @@ describe('TriggerExecutionContextFactory', () => {
 				'update',
 				expect.any(Function),
 				fence,
+				undefined,
+			);
+		});
+
+		test('threads a prefetched cursor through to getExecutePollFunctions', async () => {
+			const workflowData = buildWorkflowData();
+			const additionalData = mock<IWorkflowExecuteAdditionalData>();
+			vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData);
+
+			const pollFunctions = mock<IPollFunctions>();
+			const getPollFunctions = vi.fn().mockReturnValue(pollFunctions);
+			const getExecutePollFunctionsSpy = vi
+				.spyOn(factory, 'getExecutePollFunctions')
+				.mockReturnValue(getPollFunctions as unknown as IGetExecutePollFunctions);
+			const fence = { taskId: 'task-1', leaseEpoch: 3 };
+			const prefetched = { lastItemId: 'prefetched' };
+
+			await factory.createPollExecutionContext(workflowData, pollNode, fence, prefetched);
+
+			expect(getExecutePollFunctionsSpy).toHaveBeenCalledWith(
+				workflowData,
+				additionalData,
+				'trigger',
+				'update',
+				expect.any(Function),
+				fence,
+				prefetched,
 			);
 		});
 

packages/cli/src/workflows/triggers/poll-backoff.service.ts

@@ -1,6 +1,6 @@
 import { Logger } from '@n8n/backend-common';
 import { SchedulerConfig, WorkflowsConfig } from '@n8n/config';
-import type { PollerFailureState } from '@n8n/db';
+import type { PollerFailureState, PollerFullState } from '@n8n/db';
 import { PollerStateRepository } from '@n8n/db';
 import { Service } from '@n8n/di';
 import { ErrorReporter } from 'n8n-core';
@@ -37,17 +37,17 @@ export class PollBackoffService {
 	}
 
 	/**
-	 * Reads the stored failure state.
+	 * Reads the stored poller state: the cursor plus the failure counters.
 	 *
 	 * Never throws: `null` means no stored state, feature disabled, or a failed read.
 	 */
-	async getFailureState(workflowId: string, nodeId: string): Promise<PollerFailureState | null> {
+	async getState(workflowId: string, nodeId: string): Promise<PollerFullState | null> {
 		if (!this.enabled) return null;
 
 		try {
-			return await this.pollerStateRepository.findFailureState(workflowId, nodeId);
+			return await this.pollerStateRepository.findState(workflowId, nodeId);
 		} catch (error) {
-			this.reportFailure(error, workflowId, nodeId, 'Failed to read poller failure state');
+			this.reportFailure(error, workflowId, nodeId, 'Failed to read poller state');
 			return null;
 		}
 	}

packages/cli/src/workflows/triggers/poll-cursor.service.ts

@@ -92,6 +92,10 @@ export class PollCursorService {
 	 * @param nodeId - Poll trigger node to resolve the cursor for.
 	 * @param nodeStaticData - Node's current cursor value, used to seed the new
 	 *   storage the first time this node migrates.
+	 * @param prefetchedCursor - Cursor the task handler already read at the start
+	 *   of the tick. When present, it is returned as-is and the row is not read
+	 *   again. When absent, the read-or-insert path runs as before: nothing was
+	 *   prefetched, or the row does not exist yet.
 	 * @returns The cursor to use if this node is on the new storage, otherwise
 	 *   `{ migrated: false }` to keep using the node's own static data.
 	 */
@@ -99,6 +103,7 @@ export class PollCursorService {
 		workflowId: string,
 		nodeId: string,
 		nodeStaticData: PollCursor,
+		prefetchedCursor?: PollerCursor,
 	): Promise<{ migrated: true; cursor: PollCursor } | { migrated: false }> {
 		if (!this.enabled) {
 			const existing = await this.pollerStateRepository.findCursor(workflowId, nodeId);
@@ -108,6 +113,10 @@ export class PollCursorService {
 			return { migrated: true, cursor: toPollCursor(existing) };
 		}
 
+		if (prefetchedCursor !== undefined) {
+			return { migrated: true, cursor: toPollCursor(prefetchedCursor) };
+		}
+
 		const stored = await this.transactionRunner.run(
 			{},
 			async (ctx) =>

packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts

@@ -1,5 +1,5 @@
 import { Logger } from '@n8n/backend-common';
-import type { IWorkflowDb, PollLeaseFence } from '@n8n/db';
+import type { IWorkflowDb, PollerCursor, PollLeaseFence } 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';
@@ -276,6 +276,7 @@ export class TriggerExecutionContextFactory {
 		// service directly and this parameter will go away.
 		resolveWorkflowData: () => Promise<IWorkflowBase>,
 		fence?: PollLeaseFence,
+		prefetchedCursor?: PollerCursor,
 	): IGetExecutePollFunctions {
 		return (workflow: Workflow, node: INode) => {
 			// A poll's staged snapshot lives in an async scope entered per poll, rather
@@ -292,6 +293,7 @@ export class TriggerExecutionContextFactory {
 					workflowData.id,
 					node.id,
 					workflow.getStaticData('node', node),
+					prefetchedCursor,
 				);
 				const store = resolved.migrated
 					? { migrated: true as const, snapshot: cloneDeep(resolved.cursor), seed: resolved.cursor }
@@ -421,6 +423,7 @@ export class TriggerExecutionContextFactory {
 		workflowData: IWorkflowBase,
 		node: INode,
 		fence?: PollLeaseFence,
+		prefetchedCursor?: PollerCursor,
 	): Promise<{ workflow: Workflow; pollFunctions: IPollFunctions }> {
 		const workflow = new Workflow({
 			id: workflowData.id,
@@ -448,6 +451,7 @@ export class TriggerExecutionContextFactory {
 			'update',
 			resolveWorkflowData,
 			fence,
+			prefetchedCursor,
 		);
 		// getPollFunctions already closed over these; its signature still requires them.
 		const pollFunctions = getPollFunctions(workflow, node, additionalData, 'trigger', 'update');

packages/cli/test/integration/database/repositories/poller-state.repository.test.ts

@@ -245,26 +245,48 @@ describe('PollerStateRepository', () => {
 		it('does not touch the stored failure counters', async () => {
 			await seed('node-1', { lastItemId: 'a' });
 			await repository.recordFailure(workflowId, 'node-1', BACKOFF_MS);
-			const before = await repository.findFailureState(workflowId, 'node-1');
+			const before = await repository.findState(workflowId, 'node-1');
 
 			await repository.advanceCursor(workflowId, 'node-1', { lastItemId: 'b' }, {});
 
-			expect(await repository.findFailureState(workflowId, 'node-1')).toEqual(before);
+			const after = await repository.findState(workflowId, 'node-1');
+			expect(after?.backoffUntil).toEqual(before?.backoffUntil);
+			expect(after?.consecutiveErrors).toEqual(before?.consecutiveErrors);
 		});
 	});
 
-	describe('findFailureState', () => {
+	describe('findState', () => {
 		it('returns null for a node that has never polled', async () => {
-			expect(await repository.findFailureState(workflowId, 'node-1')).toBeNull();
+			expect(await repository.findState(workflowId, 'node-1')).toBeNull();
 		});
 
-		it('returns the stored counters', async () => {
+		it('returns the cursor and clean failure fields for a healthy row', async () => {
+			await seed('node-1', { lastItemId: 'a' });
+
+			expect(await repository.findState(workflowId, 'node-1')).toEqual({
+				cursor: { lastItemId: 'a' },
+				consecutiveErrors: 0,
+				backoffUntil: null,
+			});
+		});
+
+		it('distinguishes an empty cursor from a missing row', async () => {
 			await seed('node-1', {});
+
+			const state = await repository.findState(workflowId, 'node-1');
+
+			expect(state).not.toBeNull();
+			expect(state?.cursor).toEqual({});
+		});
+
+		it('returns the stored failure counters alongside the cursor', async () => {
+			await seed('node-1', { lastItemId: 'a' });
 			const sentAt = Date.now();
 			await repository.recordFailure(workflowId, 'node-1', BACKOFF_MS);
 
-			const state = await repository.findFailureState(workflowId, 'node-1');
+			const state = await repository.findState(workflowId, 'node-1');
 
+			expect(state?.cursor).toEqual({ lastItemId: 'a' });
 			expect(state?.consecutiveErrors).toBe(1);
 			expectDeadlineNear(state?.backoffUntil ?? null, sentAt, BACKOFF_MS);
 		});
@@ -279,7 +301,7 @@ describe('PollerStateRepository', () => {
 				repository.recordFailure(workflowId, 'node-1', BACKOFF_MS),
 			]);
 
-			const state = await repository.findFailureState(workflowId, 'node-1');
+			const state = await repository.findState(workflowId, 'node-1');
 			expect(state?.consecutiveErrors).toBe(2);
 		});
 
@@ -289,7 +311,7 @@ describe('PollerStateRepository', () => {
 
 			await repository.recordFailure(workflowId, 'node-1', BACKOFF_MS);
 
-			const state = await repository.findFailureState(workflowId, 'node-1');
+			const state = await repository.findState(workflowId, 'node-1');
 			expect(state?.consecutiveErrors).toBe(1);
 			expectDeadlineNear(state?.backoffUntil ?? null, sentAt, BACKOFF_MS);
 		});
@@ -319,7 +341,7 @@ describe('PollerStateRepository', () => {
 			await repository.recordFailure(workflowId, 'node-1', BACKOFF_MS);
 			await repository.recordFailure(workflowId, 'node-1', BACKOFF_MS);
 
-			const state = await repository.findFailureState(workflowId, 'node-1');
+			const state = await repository.findState(workflowId, 'node-1');
 			expect(state?.consecutiveErrors).toBe(2);
 		});
 
@@ -330,7 +352,7 @@ describe('PollerStateRepository', () => {
 
 			await repository.recordFailure(workflowId, 'node-1', 5_000);
 
-			const state = await repository.findFailureState(workflowId, 'node-1');
+			const state = await repository.findState(workflowId, 'node-1');
 			expect(state?.consecutiveErrors).toBe(2);
 			expectDeadlineNear(state?.backoffUntil ?? null, sentAt, ONE_HOUR_MS);
 		});
@@ -342,7 +364,7 @@ describe('PollerStateRepository', () => {
 			const sentAt = Date.now();
 			await repository.recordFailure(workflowId, 'node-1', ONE_HOUR_MS);
 
-			const state = await repository.findFailureState(workflowId, 'node-1');
+			const state = await repository.findState(workflowId, 'node-1');
 			expect(state?.consecutiveErrors).toBe(2);
 			expectDeadlineNear(state?.backoffUntil ?? null, sentAt, ONE_HOUR_MS);
 		});
@@ -358,7 +380,7 @@ describe('PollerStateRepository', () => {
 				await repository.recordFailure(workflowId, 'node-1', BACKOFF_MS, ctx);
 			});
 
-			const state = await repository.findFailureState(workflowId, 'node-1');
+			const state = await repository.findState(workflowId, 'node-1');
 			expect(state?.consecutiveErrors).toBe(1);
 		});
 
@@ -372,9 +394,10 @@ describe('PollerStateRepository', () => {
 				}),
 			).rejects.toThrow('execution insert failed');
 
-			expect(await repository.findFailureState(workflowId, 'node-1')).toEqual({
+			expect(await repository.findState(workflowId, 'node-1')).toEqual({
 				consecutiveErrors: 0,
 				backoffUntil: null,
+				cursor: {},
 			});
 		});
 	});
@@ -386,9 +409,10 @@ describe('PollerStateRepository', () => {
 
 			await repository.clearFailures(workflowId, 'node-1');
 
-			expect(await repository.findFailureState(workflowId, 'node-1')).toEqual({
+			expect(await repository.findState(workflowId, 'node-1')).toEqual({
 				consecutiveErrors: 0,
 				backoffUntil: null,
+				cursor: {},
 			});
 		});
 

packages/cli/test/integration/scheduling/poll-trigger-job-registrar.test.ts

@@ -80,7 +80,7 @@ describe('PollTriggerJobRegistrar', () => {
 		await pollerStateRepository.insert({ workflowId: workflow.id, nodeId: node.id, cursor: {} });
 		await pollerStateRepository.recordFailure(workflow.id, node.id, 60 * 60 * 1000);
 
-		const failingBefore = await pollerStateRepository.findFailureState(workflow.id, node.id);
+		const failingBefore = await pollerStateRepository.findState(workflow.id, node.id);
 		expect(failingBefore?.consecutiveErrors).toBe(1);
 		expect(failingBefore?.backoffUntil).toBeInstanceOf(Date);
 
@@ -99,8 +99,8 @@ describe('PollTriggerJobRegistrar', () => {
 		});
 		expect(jobs).toHaveLength(1);
 
-		const failureStateAfter = await pollerStateRepository.findFailureState(workflow.id, node.id);
-		expect(failureStateAfter).toEqual({ consecutiveErrors: 0, backoffUntil: null });
+		const failureStateAfter = await pollerStateRepository.findState(workflow.id, node.id);
+		expect(failureStateAfter).toEqual({ consecutiveErrors: 0, backoffUntil: null, cursor: {} });
 	});
 
 	it('leaves a failing poller_state row untouched when durable cursors are disabled, even though a job is inserted', async () => {
@@ -110,7 +110,7 @@ describe('PollTriggerJobRegistrar', () => {
 
 		await pollerStateRepository.insert({ workflowId: workflow.id, nodeId: node.id, cursor: {} });
 		await pollerStateRepository.recordFailure(workflow.id, node.id, 60 * 60 * 1000);
-		const failingBefore = await pollerStateRepository.findFailureState(workflow.id, node.id);
+		const failingBefore = await pollerStateRepository.findState(workflow.id, node.id);
 
 		const { inserted } = await registrar.register(
 			workflow.id,
@@ -120,7 +120,7 @@ describe('PollTriggerJobRegistrar', () => {
 		);
 
 		expect(inserted).toBe(true);
-		const failureStateAfter = await pollerStateRepository.findFailureState(workflow.id, node.id);
+		const failureStateAfter = await pollerStateRepository.findState(workflow.id, node.id);
 		expect(failureStateAfter).toEqual(failingBefore);
 	});
 });