diff --git a/packages/cli/src/__tests__/active-workflow-manager.test.ts b/packages/cli/src/__tests__/active-workflow-manager.test.ts index 8e8baf1250c..54d7d9b637a 100644 --- a/packages/cli/src/__tests__/active-workflow-manager.test.ts +++ b/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; beforeEach(() => { vi.clearAllMocks(); + pollCursorService = mock({ 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({ id: 'project-1', name: 'Test Project' }), ); - const pollCursorService = mock({ enabled: false }); - pollCursorService.resolveCursor.mockResolvedValue({ migrated: false }); - factory = new TriggerExecutionContextFactory( rootLogger, mock(), // errorReporter diff --git a/packages/cli/src/controllers/e2e.controller.ts b/packages/cli/src/controllers/e2e.controller.ts index 67b992d22f1..c29bb7d3c09 100644 --- a/packages/cli/src/controllers/e2e.controller.ts +++ b/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 }>) { diff --git a/packages/cli/test/integration/executions/poll-cursor-atomicity.test.ts b/packages/cli/test/integration/executions/poll-cursor-atomicity.test.ts index ef06851e4cc..4680e56573d 100644 --- a/packages/cli/test/integration/executions/poll-cursor-atomicity.test.ts +++ b/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. diff --git a/packages/nodes-base/nodes/E2eTest/E2eTestPollingTrigger.node.ts b/packages/nodes-base/nodes/E2eTest/E2eTestPollingTrigger.node.ts index 72f80328d08..fa01ded9976 100644 --- a/packages/nodes-base/nodes/E2eTest/E2eTestPollingTrigger.node.ts +++ b/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)]; } } diff --git a/packages/nodes-base/nodes/E2eTest/test/E2eTestPollingTrigger.test.ts b/packages/nodes-base/nodes/E2eTest/test/E2eTestPollingTrigger.test.ts new file mode 100644 index 00000000000..181b31a4377 --- /dev/null +++ b/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; + 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>) => + result?.[0].map((item) => item.json); + + beforeEach(() => { + trigger = new E2eTestPollingTrigger(); + pollFunctions = mockDeep(); + + 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({}); + }); +}); diff --git a/packages/testing/playwright/services/api-helper.ts b/packages/testing/playwright/services/api-helper.ts index f97f534eaff..9dd9c177df8 100644 --- a/packages/testing/playwright/services/api-helper.ts +++ b/packages/testing/playwright/services/api-helper.ts @@ -242,6 +242,31 @@ export class ApiHelpers { return data.count; } + async getPollerCursor( + workflowId: string, + nodeId: string, + ): Promise | 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 | null }; + }; + return data.cursor; + } + + async clearWorkflowStaticData(workflowId: string): Promise { + 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 { const response = await this.request.post('/rest/e2e/scheduled-jobs/fire-now', { data: { workflowId, nodeId }, diff --git a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-cursor-migrated.spec.ts b/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-cursor-migrated.spec.ts new file mode 100644 index 00000000000..7410bf9977a --- /dev/null +++ b/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); + }); + }, +); diff --git a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-cursor-unmigrated.spec.ts b/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-cursor-unmigrated.spec.ts new file mode 100644 index 00000000000..c3b15f82d6c --- /dev/null +++ b/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(); + }); + }, +); diff --git a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-durable.spec.ts b/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-durable.spec.ts index aac4cfbdd2f..5f61ecd7958 100644 --- a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-durable.spec.ts +++ b/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 ({ diff --git a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-helpers.ts b/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-helpers.ts index 0fa92e5f68e..f3472fa0ca7 100644 --- a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-helpers.ts +++ b/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; +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> { + 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) { + 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, + timeoutMs = 20_000, +): Promise { + 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, + windowMs = 8_000, +): Promise { + 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 { + 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 { + await api.clearWorkflowStaticData(workflowId); + await api.fireScheduledJobsNow(workflowId, nodeId); +} diff --git a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-misfire-skip.spec.ts b/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-misfire-skip.spec.ts index 17b14f76a70..a4e6c4d6235 100644 --- a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-misfire-skip.spec.ts +++ b/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 diff --git a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-workflow.ts b/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-workflow.ts index ad362e751bb..0303ab3c5fe 100644 --- a/packages/testing/playwright/tests/e2e/scheduling/poll-trigger-workflow.ts +++ b/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) => { 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] },