diff --git a/packages/@n8n/db/src/repositories/index.ts b/packages/@n8n/db/src/repositories/index.ts index b8ed9332f66..0182d4a4e56 100644 --- a/packages/@n8n/db/src/repositories/index.ts +++ b/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'; diff --git a/packages/@n8n/db/src/repositories/poller-state.repository.ts b/packages/@n8n/db/src/repositories/poller-state.repository.ts index 93900db1bd5..ef45377ff2e 100644 --- a/packages/@n8n/db/src/repositories/poller-state.repository.ts +++ b/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 { private readonly isPostgres: boolean; @@ -151,19 +155,23 @@ export class PollerStateRepository extends BaseRepository { 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 { + ): Promise { 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, + }; } /** diff --git a/packages/cli/src/controllers/e2e.controller.ts b/packages/cli/src/controllers/e2e.controller.ts index 510299e2772..7d0df0edd14 100644 --- a/packages/cli/src/controllers/e2e.controller.ts +++ b/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 { diff --git a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts index 1d923fba253..a7b09129233 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/__tests__/poll-trigger-task-handler.test.ts @@ -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); diff --git a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts index fa9de887d09..a487016e5ec 100644 --- a/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts +++ b/packages/cli/src/scheduling/poll-trigger-node/poll-trigger-task-handler.ts @@ -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. diff --git a/packages/cli/src/workflows/triggers/__tests__/poll-backoff.service.test.ts b/packages/cli/src/workflows/triggers/__tests__/poll-backoff.service.test.ts index 6eeac9b7580..e8883502250 100644 --- a/packages/cli/src/workflows/triggers/__tests__/poll-backoff.service.test.ts +++ b/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); }); diff --git a/packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts b/packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts index 02780e7887c..66ddaff84e5 100644 --- a/packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/poll-cursor.service.test.ts @@ -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); diff --git a/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts b/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts index b9a8abf6060..cd7026cf870 100644 --- a/packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts +++ b/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, + ): RunnablePollFunctions => { const getPollFunctions = factory.getExecutePollFunctions( mock({ id: 'wf-1', name: 'Test Workflow' }), additionalData, mode, activation, async () => mock({ 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(); + vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(additionalData); + + const pollFunctions = mock(); + 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, ); }); diff --git a/packages/cli/src/workflows/triggers/poll-backoff.service.ts b/packages/cli/src/workflows/triggers/poll-backoff.service.ts index 9473589960a..a066464b18a 100644 --- a/packages/cli/src/workflows/triggers/poll-backoff.service.ts +++ b/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 { + async getState(workflowId: string, nodeId: string): Promise { 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; } } diff --git a/packages/cli/src/workflows/triggers/poll-cursor.service.ts b/packages/cli/src/workflows/triggers/poll-cursor.service.ts index 1a8f04906df..a84f6f100e4 100644 --- a/packages/cli/src/workflows/triggers/poll-cursor.service.ts +++ b/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) => diff --git a/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts b/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts index 020536fe50b..346d73a5e54 100644 --- a/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts +++ b/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, 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'); diff --git a/packages/cli/test/integration/database/repositories/poller-state.repository.test.ts b/packages/cli/test/integration/database/repositories/poller-state.repository.test.ts index a8f77fa6dd3..bb2b08bfd54 100644 --- a/packages/cli/test/integration/database/repositories/poller-state.repository.test.ts +++ b/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: {}, }); }); diff --git a/packages/cli/test/integration/scheduling/poll-trigger-job-registrar.test.ts b/packages/cli/test/integration/scheduling/poll-trigger-job-registrar.test.ts index 07c75ac35f8..7fed9a01a55 100644 --- a/packages/cli/test/integration/scheduling/poll-trigger-job-registrar.test.ts +++ b/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); }); });