test(core): Add durable scheduler poll cursors e2e tests (no-changelog) (#35343)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Lorent Lempereur <lorent.lempereur@n8n.io>

packages/cli/src/__tests__/active-workflow-manager.test.ts

@@ -28,7 +28,7 @@ import type {
 } from 'n8n-workflow';
 import { sleep } from '@n8n/utils/sleep';
 import { Workflow, WorkflowActivationError } from 'n8n-workflow';
-import { mock } from 'vitest-mock-extended';
+import { mock, type MockProxy } from 'vitest-mock-extended';
 
 import type { ActivationErrorsService } from '@/activation-errors.service';
 import { ActiveWorkflowManager } from '@/active-workflow-manager';
@@ -407,9 +407,12 @@ describe('ActiveWorkflowManager', () => {
 		let scopedLogger: Logger;
 
 		let factory: TriggerExecutionContextFactory;
+		let pollCursorService: MockProxy<PollCursorService>;
 
 		beforeEach(() => {
 			vi.clearAllMocks();
+			pollCursorService = mock<PollCursorService>({ enabled: false });
+			pollCursorService.resolveCursor.mockResolvedValue({ migrated: false });
 			workflowStaticDataService.saveStaticData.mockResolvedValue(undefined);
 			workflowExecutionService.runWorkflow.mockResolvedValue('exec-123');
 			activeWorkflowTriggers.remove.mockResolvedValue(true);
@@ -424,9 +427,6 @@ describe('ActiveWorkflowManager', () => {
 				mock<Project>({ id: 'project-1', name: 'Test Project' }),
 			);
 
-			const pollCursorService = mock<PollCursorService>({ enabled: false });
-			pollCursorService.resolveCursor.mockResolvedValue({ migrated: false });
-
 			factory = new TriggerExecutionContextFactory(
 				rootLogger,
 				mock(), // errorReporter

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

@@ -9,6 +9,7 @@ import {
 	GLOBAL_CHAT_USER_ROLE,
 	GLOBAL_MEMBER_ROLE,
 	GLOBAL_OWNER_ROLE,
+	PollerStateRepository,
 	ScheduledJobRepository,
 	SettingsRepository,
 	UserRepository,
@@ -33,6 +34,7 @@ import { Push } from '@/push';
 import { CacheService } from '@/services/cache/cache.service';
 import { FrontendService } from '@/services/frontend.service';
 import { PasswordUtility } from '@/services/password.utility';
+import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';
 
 if (!inE2ETests) {
 	Container.get(Logger).error('E2E endpoints only allowed during E2E tests');
@@ -47,6 +49,7 @@ const tablesToTruncate = [
 	'execution_entity',
 	'installed_nodes',
 	'installed_packages',
+	'poller_state',
 	'project',
 	'project_relation',
 	'role',
@@ -197,6 +200,8 @@ export class E2EController {
 		private readonly executionsConfig: ExecutionsConfig,
 		private readonly logStreamingDestinationsService: LogStreamingDestinationService,
 		private readonly scheduledJobRepository: ScheduledJobRepository,
+		private readonly pollerStateRepository: PollerStateRepository,
+		private readonly workflowStaticDataService: WorkflowStaticDataService,
 	) {
 		license.isLicensed = (feature: BooleanLicenseFeature) => this.enabledFeatures[feature] ?? false;
 
@@ -262,6 +267,28 @@ export class E2EController {
 		return { count };
 	}
 
+	/**
+	 * A poll node's stored cursor, so a test can assert on it directly instead of
+	 * inferring it from execution behaviour.
+	 */
+	@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);
+		return { cursor };
+	}
+
+	/**
+	 * Wipes a workflow's static data, the store an unmigrated poll cursor lives in.
+	 * The workflow DTOs deliberately drop `staticData` writes, so a test has no way
+	 * to reset that state through the workflow API.
+	 */
+	@Post('/workflow-static-data/clear', { skipAuth: true })
+	async clearWorkflowStaticData(req: Request<{}, {}, { workflowId: string }>) {
+		await this.workflowStaticDataService.saveStaticDataById(req.body.workflowId, {});
+		return { success: true };
+	}
+
 	/** Lets a test observe a real scheduled dispatch without waiting out the job's cron interval. */
 	@Post('/scheduled-jobs/fire-now', { skipAuth: true })
 	async fireScheduledJobsNow(req: Request<{}, {}, { workflowId: string; nodeId: string }>) {

packages/cli/test/integration/executions/poll-cursor-atomicity.test.ts

@@ -148,6 +148,41 @@ describe('poll cursor atomicity', () => {
 		expect(await pollerStateRepository.findCursor(workflow.id, nodeId)).toBeNull();
 	});
 
+	it('seeds the row from the static-data cursor already accrued while the flag was off, rather than resetting it', async () => {
+		pollerConfig.durableCursorsEnabled = false;
+
+		// Simulate several polls accruing a cursor in the node's static data while
+		// durable cursors are off: resolveCursor reports `migrated: false` and never
+		// touches poller_state, mirroring the unmigrated path e2e-tested in
+		// poll-trigger-cursor-unmigrated.spec.ts.
+		for (const lastItemId of ['a', 'b', 'c']) {
+			expect(await pollCursorService.resolveCursor(workflow.id, nodeId, { lastItemId })).toEqual({
+				migrated: false,
+			});
+		}
+		expect(await pollerStateRepository.findCursor(workflow.id, nodeId)).toBeNull();
+
+		// Flip the flag on for this already-running workflow and poll again, passing
+		// through the cursor its static data accrued while the flag was off.
+		pollerConfig.durableCursorsEnabled = true;
+		const migrated = await pollCursorService.resolveCursor(workflow.id, nodeId, {
+			lastItemId: 'c',
+		});
+
+		// The row resumes forward from the accrued value - it is not seeded null, which
+		// would otherwise re-emit every item the workflow already saw as unmigrated.
+		expect(migrated).toEqual({ migrated: true, cursor: { lastItemId: 'c' } });
+		expect(await pollerStateRepository.findCursor(workflow.id, nodeId)).toEqual({
+			lastItemId: 'c',
+		});
+
+		// Once migrated, the row is the source of truth: a later poll's static-data
+		// blob no longer has any effect, even if it disagrees with the stored cursor.
+		expect(
+			await pollCursorService.resolveCursor(workflow.id, nodeId, { lastItemId: 'stale' }),
+		).toEqual({ migrated: true, cursor: { lastItemId: 'c' } });
+	});
+
 	it('still advances the cursor when the execution insert fails and the flag is off, for a node that already migrated', async () => {
 		// Migrate the row while the flag is on, then flip it off: reads still prefer
 		// the row, and the flag only narrows to write atomicity from here on.

packages/nodes-base/nodes/E2eTest/E2eTestPollingTrigger.node.ts

@@ -12,6 +12,12 @@ interface PollResponseBody {
 	items?: IDataObject[];
 }
 
+const highestItemId = (items: IDataObject[], startingFrom: number) =>
+	items.reduce((highest, item) => {
+		const id = Number(item.id);
+		return Number.isFinite(id) && id > highest ? id : highest;
+	}, startingFrom);
+
 export class E2eTestPollingTrigger implements INodeType {
 	description: INodeTypeDescription = {
 		displayName: 'E2E Test Polling Trigger',
@@ -53,8 +59,23 @@ export class E2eTestPollingTrigger implements INodeType {
 			throw new NodeOperationError(this.getNode(), error as Error);
 		}
 
-		if (!body.items?.length) return null;
+		const items = body.items ?? [];
+		if (items.length === 0) return null;
 
-		return [this.helpers.returnJsonArray(body.items)];
+		const nodeStaticData = this.getWorkflowStaticData('node');
+		const lastItemId =
+			typeof nodeStaticData.lastItemId === 'number' ? nodeStaticData.lastItemId : null;
+
+		const newItems = items.filter((item) => {
+			const id = Number(item.id);
+			return Number.isFinite(id) && (lastItemId === null || id > lastItemId);
+		});
+
+		// Set even when nothing new is emitted below, so the advance still persists.
+		nodeStaticData.lastItemId = highestItemId(items, lastItemId ?? 0);
+
+		if (newItems.length === 0) return null;
+
+		return [this.helpers.returnJsonArray(newItems)];
 	}
 }

packages/nodes-base/nodes/E2eTest/test/E2eTestPollingTrigger.test.ts

@@ -0,0 +1,121 @@
+import type { IDataObject, INode, IPollFunctions, PollCursor } from 'n8n-workflow';
+import { NodeOperationError } from 'n8n-workflow';
+import type { Mock, Mocked } from 'vitest';
+import { mockDeep } from 'vitest-mock-extended';
+
+import { E2eTestPollingTrigger } from '../E2eTestPollingTrigger.node';
+
+describe('E2eTestPollingTrigger', () => {
+	const node: INode = {
+		id: 'poll-node-id',
+		name: 'E2E Test Polling Trigger',
+		type: 'n8n-nodes-base.e2eTestPollingTrigger',
+		typeVersion: 1,
+		position: [0, 0],
+		parameters: {},
+	};
+
+	let trigger: E2eTestPollingTrigger;
+	let pollFunctions: Mocked<IPollFunctions>;
+	let nodeStaticData: PollCursor;
+
+	const givenCursor = (cursor: PollCursor | null) => {
+		nodeStaticData = cursor ? { ...cursor } : {};
+	};
+
+	const givenResponse = (body: unknown) => {
+		(pollFunctions.helpers.httpRequest as Mock).mockResolvedValue(body);
+	};
+
+	const emittedJson = (result: Awaited<ReturnType<E2eTestPollingTrigger['poll']>>) =>
+		result?.[0].map((item) => item.json);
+
+	beforeEach(() => {
+		trigger = new E2eTestPollingTrigger();
+		pollFunctions = mockDeep<IPollFunctions>();
+
+		pollFunctions.getNode.mockReturnValue(node);
+		pollFunctions.getNodeParameter.mockReturnValue('http://poll.test/items');
+		(pollFunctions.helpers.returnJsonArray as Mock).mockImplementation((data: IDataObject[]) =>
+			data.map((json, index) => ({ json, pairedItem: { item: index } })),
+		);
+		givenCursor(null);
+		pollFunctions.getWorkflowStaticData.mockImplementation(() => nodeStaticData);
+	});
+
+	it('should emit every item and set the highest id on its first poll', async () => {
+		givenResponse({ items: [{ id: 1 }, { id: 2 }] });
+
+		const result = await trigger.poll.call(pollFunctions);
+
+		expect(emittedJson(result)).toEqual([{ id: 1 }, { id: 2 }]);
+		expect(nodeStaticData).toEqual({ lastItemId: 2 });
+		expect(pollFunctions.getWorkflowStaticData).toHaveBeenCalledWith('node');
+	});
+
+	it('should never emit an item with a non-numeric or missing id, on a first poll', async () => {
+		givenResponse({ items: [{ id: 1 }, { id: 'not-a-number' }, { foo: 'bar' }] });
+
+		const result = await trigger.poll.call(pollFunctions);
+
+		expect(emittedJson(result)).toEqual([{ id: 1 }]);
+		expect(nodeStaticData).toEqual({ lastItemId: 1 });
+	});
+
+	it('should never emit an item with a non-numeric or missing id, on a later poll', async () => {
+		givenCursor({ lastItemId: 2 });
+		givenResponse({ items: [{ id: 3 }, { id: 'not-a-number' }, { foo: 'bar' }] });
+
+		const result = await trigger.poll.call(pollFunctions);
+
+		expect(emittedJson(result)).toEqual([{ id: 3 }]);
+		expect(nodeStaticData).toEqual({ lastItemId: 3 });
+	});
+
+	it('should emit only the items above the cursor and advance it', async () => {
+		givenCursor({ lastItemId: 2 });
+		givenResponse({ items: [{ id: 1 }, { id: 2 }, { id: 3 }, { id: 4 }] });
+
+		const result = await trigger.poll.call(pollFunctions);
+
+		expect(emittedJson(result)).toEqual([{ id: 3 }, { id: 4 }]);
+		expect(nodeStaticData).toEqual({ lastItemId: 4 });
+	});
+
+	it('should emit nothing and hold the cursor when the endpoint repeats known items', async () => {
+		givenCursor({ lastItemId: 2 });
+		givenResponse({ items: [{ id: 1 }, { id: 2 }] });
+
+		const result = await trigger.poll.call(pollFunctions);
+
+		expect(result).toBeNull();
+		expect(nodeStaticData).toEqual({ lastItemId: 2 });
+	});
+
+	it('should emit nothing and leave the cursor untouched when the endpoint returns no items', async () => {
+		givenCursor({ lastItemId: 2 });
+		givenResponse({ items: [] });
+
+		const result = await trigger.poll.call(pollFunctions);
+
+		expect(result).toBeNull();
+		expect(nodeStaticData).toEqual({ lastItemId: 2 });
+	});
+
+	it('should treat an unusable cursor value as a first poll', async () => {
+		givenCursor({ lastItemId: 'not-a-number' });
+		givenResponse({ items: [{ id: 7 }] });
+
+		const result = await trigger.poll.call(pollFunctions);
+
+		expect(emittedJson(result)).toEqual([{ id: 7 }]);
+		expect(nodeStaticData).toEqual({ lastItemId: 7 });
+	});
+
+	it('should raise a node operation error when the endpoint fails', async () => {
+		(pollFunctions.helpers.httpRequest as Mock).mockRejectedValue(new Error('connection refused'));
+
+		await expect(trigger.poll.call(pollFunctions)).rejects.toThrow(NodeOperationError);
+		expect(nodeStaticData).toEqual({});
+	});
+});

packages/testing/playwright/services/api-helper.ts

@@ -242,6 +242,31 @@ export class ApiHelpers {
 		return data.count;
 	}
 
+	async getPollerCursor(
+		workflowId: string,
+		nodeId: string,
+	): Promise<Record<string, unknown> | null> {
+		const response = await this.request.get('/rest/e2e/poller-state', {
+			params: { workflowId, nodeId },
+		});
+		if (!response.ok()) {
+			throw new TestError(`Failed to get poller cursor: ${await response.text()}`);
+		}
+		const { data } = (await response.json()) as {
+			data: { cursor: Record<string, unknown> | null };
+		};
+		return data.cursor;
+	}
+
+	async clearWorkflowStaticData(workflowId: string): Promise<void> {
+		const response = await this.request.post('/rest/e2e/workflow-static-data/clear', {
+			data: { workflowId },
+		});
+		if (!response.ok()) {
+			throw new TestError(`Failed to clear workflow static data: ${await response.text()}`);
+		}
+	}
+
 	async fireScheduledJobsNow(workflowId: string, nodeId: string): Promise<void> {
 		const response = await this.request.post('/rest/e2e/scheduled-jobs/fire-now', {
 			data: { workflowId, nodeId },

packages/testing/playwright/tests/e2e/scheduling/poll-trigger-cursor-migrated.spec.ts

@@ -0,0 +1,119 @@
+import {
+	clearStaticDataAndPoll,
+	expectNewTriggerExecution,
+	expectNoNewTriggerExecution,
+	expectPollTriggerFires,
+	readNodeStaticData,
+	fetchTriggerExecutionIds,
+} from './poll-trigger-helpers';
+import { makePollTriggerWorkflow, POLL_TRIGGER_NODE_NAME } from './poll-trigger-workflow';
+import { test, expect } from '../../../fixtures/base';
+
+test.use({
+	capability: {
+		services: ['proxy'],
+		env: {
+			N8N_POLLER_DURABLE_CURSORS_ENABLED: 'true',
+			N8N_SCHEDULER_ENABLED: 'true',
+			N8N_USE_WORKFLOW_PUBLICATION_SERVICE: 'true',
+			N8N_SCHEDULER_POLL_TRIGGERS_ENABLED: 'true',
+			N8N_SCHEDULER_MATERIALIZATION_INTERVAL: '1',
+			N8N_SCHEDULER_EXECUTOR_INTERVAL: '1',
+		},
+	},
+});
+
+test.describe(
+	'Poll Trigger cursor (migrated) @capability:proxy',
+	{
+		annotation: [{ type: 'owner', description: 'Catalysts' }],
+	},
+	() => {
+		test('should emit an item past the cursor and advance it in poller_state', async ({
+			api,
+			services,
+		}) => {
+			const { workflowId, nodeId } = await expectPollTriggerFires(
+				api,
+				services.proxy,
+				makePollTriggerWorkflow,
+				{ itemsAfterSeedPoll: [{ id: 1 }, { id: 2 }] },
+			);
+
+			await expect
+				.poll(async () => await api.getPollerCursor(workflowId, nodeId), { timeout: 15_000 })
+				.toEqual({ lastItemId: 1 });
+
+			const afterSeedPoll = await fetchTriggerExecutionIds(api, workflowId);
+			await api.fireScheduledJobsNow(workflowId, nodeId);
+
+			await expectNewTriggerExecution(api, workflowId, afterSeedPoll);
+
+			await expect
+				.poll(async () => await api.getPollerCursor(workflowId, nodeId), { timeout: 15_000 })
+				.toEqual({ lastItemId: 2 });
+		});
+
+		test('should keep the cursor when the workflow static data is cleared', async ({
+			api,
+			services,
+		}) => {
+			const { workflowId, nodeId } = await expectPollTriggerFires(
+				api,
+				services.proxy,
+				makePollTriggerWorkflow,
+			);
+
+			const afterSeedPoll = await fetchTriggerExecutionIds(api, workflowId);
+			await clearStaticDataAndPoll(api, workflowId, nodeId);
+
+			await expectNoNewTriggerExecution(api, workflowId, afterSeedPoll);
+			expect(await readNodeStaticData(api, workflowId, POLL_TRIGGER_NODE_NAME)).toBeNull();
+			expect(await api.getPollerCursor(workflowId, nodeId)).toEqual({ lastItemId: 1 });
+		});
+
+		// `fireScheduledJobsNow` backdates the job's `nextRunAt` without waiting for the
+		// poll to run, so firing it twice back-to-back is the closest this gets to racing
+		// two cursor commits; the scheduler still serializes which pass claims the job, so
+		// the two `advanceCursor` writes never actually interleave. What it proves: neither
+		// tick's item is dropped, and the cursor lands on the higher id, not an
+		// intermediate value.
+		test('should not lose either poll when two ticks are fired back-to-back', async ({
+			api,
+			services,
+		}) => {
+			// The item both ticks below will race to report as new: registered as the one
+			// unlimited wave after the seed, since MockServer expectations only stay ordered
+			// while the earlier one is one-shot.
+			const { workflowId, nodeId } = await expectPollTriggerFires(
+				api,
+				services.proxy,
+				makePollTriggerWorkflow,
+				{ itemsAfterSeedPoll: [{ id: 2 }] },
+			);
+
+			await expect
+				.poll(async () => await api.getPollerCursor(workflowId, nodeId), { timeout: 15_000 })
+				.toEqual({ lastItemId: 1 });
+
+			const afterSeedPoll = await fetchTriggerExecutionIds(api, workflowId);
+
+			await Promise.all([
+				api.fireScheduledJobsNow(workflowId, nodeId),
+				api.fireScheduledJobsNow(workflowId, nodeId),
+			]);
+
+			await expect
+				.poll(async () => await api.getPollerCursor(workflowId, nodeId), { timeout: 20_000 })
+				.toEqual({ lastItemId: 2 });
+
+			// Only one of the two concurrent ticks should have found item 2 new; the other
+			// must see it already reflected in the cursor and emit nothing.
+			await expect
+				.poll(async () => (await fetchTriggerExecutionIds(api, workflowId)).size, {
+					timeout: 20_000,
+				})
+				.toBe(afterSeedPoll.size + 1);
+		});
+	},
+);

packages/testing/playwright/tests/e2e/scheduling/poll-trigger-cursor-unmigrated.spec.ts

@@ -0,0 +1,77 @@
+import {
+	clearStaticDataAndPoll,
+	expectNewTriggerExecution,
+	expectPollTriggerFires,
+	readNodeStaticData,
+	fetchTriggerExecutionIds,
+} from './poll-trigger-helpers';
+import { makePollTriggerWorkflow, POLL_TRIGGER_NODE_NAME } from './poll-trigger-workflow';
+import { test, expect } from '../../../fixtures/base';
+
+test.use({
+	capability: {
+		services: ['proxy'],
+		env: {
+			N8N_SCHEDULER_ENABLED: 'true',
+			N8N_USE_WORKFLOW_PUBLICATION_SERVICE: 'true',
+			N8N_SCHEDULER_POLL_TRIGGERS_ENABLED: 'true',
+			N8N_SCHEDULER_MATERIALIZATION_INTERVAL: '1',
+			N8N_SCHEDULER_EXECUTOR_INTERVAL: '1',
+		},
+	},
+});
+
+test.describe(
+	'Poll Trigger cursor (unmigrated) @capability:proxy',
+	{
+		annotation: [{ type: 'owner', description: 'Catalysts' }],
+	},
+	() => {
+		test('should advance the cursor in the workflow static data', async ({ api, services }) => {
+			const { workflowId, nodeId } = await expectPollTriggerFires(
+				api,
+				services.proxy,
+				makePollTriggerWorkflow,
+				{ itemsAfterSeedPoll: [{ id: 1 }, { id: 2 }] },
+			);
+
+			await expect
+				.poll(async () => await readNodeStaticData(api, workflowId, POLL_TRIGGER_NODE_NAME), {
+					timeout: 15_000,
+				})
+				.toEqual({ lastItemId: 1 });
+			expect(await api.getPollerCursor(workflowId, nodeId)).toBeNull();
+
+			const afterSeedPoll = await fetchTriggerExecutionIds(api, workflowId);
+			await api.fireScheduledJobsNow(workflowId, nodeId);
+			await expectNewTriggerExecution(api, workflowId, afterSeedPoll);
+
+			await expect
+				.poll(async () => await readNodeStaticData(api, workflowId, POLL_TRIGGER_NODE_NAME), {
+					timeout: 15_000,
+				})
+				.toEqual({ lastItemId: 2 });
+			expect(await api.getPollerCursor(workflowId, nodeId)).toBeNull();
+		});
+
+		test('should restart from the first item when the workflow static data is cleared', async ({
+			api,
+			services,
+		}) => {
+			const { workflowId, nodeId } = await expectPollTriggerFires(
+				api,
+				services.proxy,
+				makePollTriggerWorkflow,
+			);
+
+			const afterSeedPoll = await fetchTriggerExecutionIds(api, workflowId);
+			await clearStaticDataAndPoll(api, workflowId, nodeId);
+
+			await expectNewTriggerExecution(api, workflowId, afterSeedPoll);
+			expect(await readNodeStaticData(api, workflowId, POLL_TRIGGER_NODE_NAME)).toEqual({
+				lastItemId: 1,
+			});
+			expect(await api.getPollerCursor(workflowId, nodeId)).toBeNull();
+		});
+	},
+);

packages/testing/playwright/tests/e2e/scheduling/poll-trigger-durable.spec.ts

@@ -1,4 +1,8 @@
-import { expectPollTriggerFires } from './poll-trigger-helpers';
+import {
+	expectNewTriggerExecution,
+	expectPollTriggerFires,
+	fetchTriggerExecutionIds,
+} from './poll-trigger-helpers';
 import { makePollTriggerWorkflow, makeCronPollTriggerWorkflow } from './poll-trigger-workflow';
 import { test, expect } from '../../../fixtures/base';
 
@@ -49,21 +53,17 @@ test.describe(
 			// `fireScheduledJobsNow` forces the job's `nextRunAt` to now so the 1s
 			// sweep configured above claims it, instead of waiting out the real
 			// cron interval.
-			const { workflowId, nodeId, path } = await expectPollTriggerFires(
+			const { workflowId, nodeId } = await expectPollTriggerFires(
 				api,
 				services.proxy,
 				makePollTriggerWorkflow,
+				{ itemsAfterSeedPoll: [{ id: 1 }, { id: 2 }] },
 			);
 
-			await services.proxy.createGetExpectation(path, { items: [{ id: 2 }] });
+			const afterSeedPoll = await fetchTriggerExecutionIds(api, workflowId);
 			await api.fireScheduledJobsNow(workflowId, nodeId);
 
-			const scheduledExecution = await api.workflows.waitForExecution(
-				workflowId,
-				15_000,
-				'trigger',
-			);
-			expect(scheduledExecution.status).toBe('success');
+			await expectNewTriggerExecution(api, workflowId, afterSeedPoll);
 		});
 
 		test('should remove the durable job when the workflow is deactivated', async ({

packages/testing/playwright/tests/e2e/scheduling/poll-trigger-helpers.ts

@@ -1,5 +1,5 @@
 import type { ProxyServer } from 'n8n-containers/services/proxy';
-import type { IWorkflowBase } from 'n8n-workflow';
+import type { IDataObject, IWorkflowBase } from 'n8n-workflow';
 import { nanoid } from 'nanoid';
 
 import type { makePollTriggerWorkflow } from './poll-trigger-workflow';
@@ -8,15 +8,43 @@ import type { ApiHelpers } from '../../../services/api-helper';
 
 type PollTriggerWorkflow = ReturnType<typeof makePollTriggerWorkflow>;
 
+const SEED_POLL_ITEMS: IDataObject[] = [{ id: 1 }];
+
+export async function programPollResponse(
+	proxy: ProxyServer,
+	path: string,
+	items: IDataObject[],
+	times?: { remainingTimes: number; unlimited: boolean },
+) {
+	await proxy.createExpectation({
+		httpRequest: { method: 'GET', path },
+		httpResponse: {
+			statusCode: 200,
+			headers: { 'Content-Type': ['application/json'] },
+			body: JSON.stringify({ items }),
+		},
+		times,
+	});
+}
+
 // Programs the mock poll response before activation, so the inline seed poll
 // that every fresh activation runs is itself the fire under test.
 export async function expectPollTriggerFires(
 	api: ApiHelpers,
 	proxy: ProxyServer,
 	makeWorkflow: (path: string) => PollTriggerWorkflow,
+	options?: { itemsAfterSeedPoll?: IDataObject[] },
 ): Promise<{ workflowId: string; nodeId: string; path: string }> {
 	const path = `/${nanoid()}`;
-	await proxy.createGetExpectation(path, { items: [{ id: 1 }] });
+	const { itemsAfterSeedPoll } = options ?? {};
+
+	await programPollResponse(
+		proxy,
+		path,
+		SEED_POLL_ITEMS,
+		itemsAfterSeedPoll && { remainingTimes: 1, unlimited: false },
+	);
+	if (itemsAfterSeedPoll) await programPollResponse(proxy, path, itemsAfterSeedPoll);
 
 	const { workflowId, createdWorkflow } = await api.workflows.createWorkflowFromDefinition(
 		makeWorkflow(path).toJSON() as IWorkflowBase,
@@ -34,3 +62,86 @@ export async function expectPollTriggerFires(
 
 	return { workflowId, nodeId: triggerNode.id, path };
 }
+
+export async function fetchTriggerExecutionIds(
+	api: ApiHelpers,
+	workflowId: string,
+): Promise<Set<string>> {
+	const executions = await api.workflows.getExecutions(workflowId, 50);
+	return new Set(
+		executions.filter((execution) => execution.mode === 'trigger').map((execution) => execution.id),
+	);
+}
+
+async function fetchNewTriggerExecutions(api: ApiHelpers, workflowId: string, known: Set<string>) {
+	const executions = await api.workflows.getExecutions(workflowId, 50);
+	return executions.filter((execution) => execution.mode === 'trigger' && !known.has(execution.id));
+}
+
+// Only an execution whose id is absent from `known` proves a fresh fire;
+// `waitForExecution`'s recency fallback would otherwise re-match the
+// activation-seed execution. Requires the count to repeat across polls,
+// with all executions already 'success', before treating it as settled
+// — catches a duplicate re-emit from a racing cursor commit.
+export async function expectNewTriggerExecution(
+	api: ApiHelpers,
+	workflowId: string,
+	known: Set<string>,
+	timeoutMs = 20_000,
+): Promise<void> {
+	let previousCount = -1;
+
+	await expect
+		.poll(
+			async () => {
+				const fresh = await fetchNewTriggerExecutions(api, workflowId, known);
+				const count = fresh.length;
+				const settled =
+					count > 0 &&
+					count === previousCount &&
+					fresh.every((execution) => execution.status === 'success');
+				previousCount = count;
+				return settled;
+			},
+			{ timeout: timeoutMs },
+		)
+		.toBe(true);
+
+	const fresh = await fetchNewTriggerExecutions(api, workflowId, known);
+
+	expect(fresh).toHaveLength(1);
+	expect(fresh[0].status).toBe('success');
+}
+
+export async function expectNoNewTriggerExecution(
+	api: ApiHelpers,
+	workflowId: string,
+	known: Set<string>,
+	windowMs = 8_000,
+): Promise<void> {
+	await new Promise((resolve) => setTimeout(resolve, windowMs));
+	expect(await fetchNewTriggerExecutions(api, workflowId, known)).toHaveLength(0);
+}
+
+export async function readNodeStaticData(
+	api: ApiHelpers,
+	workflowId: string,
+	nodeName: string,
+): Promise<unknown> {
+	const { staticData } = await api.workflows.getWorkflow(workflowId);
+	const parsed =
+		typeof staticData === 'string' ? (JSON.parse(staticData) as IDataObject) : staticData;
+	return parsed?.[`node:${nodeName}`] ?? null;
+}
+
+// Wipes the static data and forces the poll that reads it back. The workflow stays
+// published: every scheduled poll re-reads the static data from the workflow row,
+// so a deactivate/reactivate cycle would add nothing but timing.
+export async function clearStaticDataAndPoll(
+	api: ApiHelpers,
+	workflowId: string,
+	nodeId: string,
+): Promise<void> {
+	await api.clearWorkflowStaticData(workflowId);
+	await api.fireScheduledJobsNow(workflowId, nodeId);
+}

packages/testing/playwright/tests/e2e/scheduling/poll-trigger-misfire-skip.spec.ts

@@ -26,8 +26,11 @@ test.describe(
 		test('drops a missed backlog instead of firing a stale poll', async ({ api, services }) => {
 			// Six hours between ticks: whatever fires in this test's short observation
 			// window can only be the backdated backlog, never the schedule's own next tick.
-			const { workflowId, nodeId } = await expectPollTriggerFires(api, services.proxy, (path) =>
-				makeCronPollTriggerWorkflow(path, '0 0 */6 * * *'),
+			const { workflowId, nodeId } = await expectPollTriggerFires(
+				api,
+				services.proxy,
+				(path) => makeCronPollTriggerWorkflow(path, '0 0 */6 * * *'),
+				{ itemsAfterSeedPoll: [{ id: 1 }, { id: 2 }] },
 			);
 
 			// A day of backlog, all past the 3s grace: a materializer walking the plain

packages/testing/playwright/tests/e2e/scheduling/poll-trigger-workflow.ts

@@ -1,12 +1,14 @@
 import { workflow, trigger, node } from '@n8n/workflow-sdk';
 import { nanoid } from 'nanoid';
 
+export const POLL_TRIGGER_NODE_NAME = 'E2E Test Polling Trigger';
+
 const buildWorkflow = (path: string, pollTimesItem: Record<string, unknown>) => {
 	const pollTrigger = trigger({
 		type: 'n8n-nodes-base.e2eTestPollingTrigger',
 		version: 1,
 		config: {
-			name: 'E2E Test Polling Trigger',
+			name: POLL_TRIGGER_NODE_NAME,
 			parameters: {
 				url: `http://e2e-poll-test.local${path}`,
 				pollTimes: { item: [pollTimesItem] },