diff --git a/packages/cli/src/services/__tests__/engine-v2-payload-guard.service.test.ts b/packages/cli/src/services/__tests__/engine-v2-payload-guard.service.test.ts index 642f0f9930e..35852188ce8 100644 --- a/packages/cli/src/services/__tests__/engine-v2-payload-guard.service.test.ts +++ b/packages/cli/src/services/__tests__/engine-v2-payload-guard.service.test.ts @@ -34,21 +34,21 @@ describe('EngineV2PayloadGuard', () => { { name: 'a null slot', slots: [null] }, { name: 'items with no binary key', slots: [[{ json: {} }]] }, { name: 'an empty binary map', slots: [[{ json: {}, binary: {} }]] }, - ])('allows a payload with $name', async ({ slots }) => { - await expect(guard.assertNoFiles(slots, REASON)).resolves.toBeUndefined(); + ])('allows a payload with $name', ({ slots }) => { + expect(() => guard.assertNoFiles(slots, REASON)).not.toThrow(); expect(binaryDataService.deleteManyByBinaryDataId).not.toHaveBeenCalled(); }); - it('refuses a stored file and deletes it, because no execution will own it', async () => { + it('refuses a stored file and deletes it, because no execution will own it', () => { const slots = [[{ json: {}, binary: { attachment: stored('filesystem:abc') } }]]; - await expect(guard.assertNoFiles(slots, REASON)).rejects.toThrow(REASON); + expect(() => guard.assertNoFiles(slots, REASON)).toThrow(REASON); expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledExactlyOnceWith([ 'filesystem:abc', ]); }); - it('deletes every stored file across slots and items', async () => { + it('deletes every stored file across slots and items', () => { const slots: Array = [ [ { json: {}, binary: { a: stored('filesystem:1') } }, @@ -58,7 +58,7 @@ describe('EngineV2PayloadGuard', () => { [{ json: {}, binary: { d: stored('filesystem:3') } }], ]; - await expect(guard.assertNoFiles(slots, REASON)).rejects.toThrow(UserError); + expect(() => guard.assertNoFiles(slots, REASON)).toThrow(UserError); expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledExactlyOnceWith([ 'filesystem:1', 'filesystem:2', @@ -66,10 +66,10 @@ describe('EngineV2PayloadGuard', () => { ]); }); - it('refuses an in-memory file without a delete, as it has no id', async () => { + it('refuses an in-memory file without a delete, as it has no id', () => { const slots = [[{ json: {}, binary: { attachment: inMemory() } }]]; - await expect(guard.assertNoFiles(slots, REASON)).rejects.toThrow(REASON); + expect(() => guard.assertNoFiles(slots, REASON)).toThrow(REASON); expect(binaryDataService.deleteManyByBinaryDataId).not.toHaveBeenCalled(); }); @@ -77,35 +77,13 @@ describe('EngineV2PayloadGuard', () => { binaryDataService.deleteManyByBinaryDataId.mockRejectedValue(new Error('store is down')); const slots = [[{ json: {}, binary: { attachment: stored('s3:abc') } }]]; - await expect(guard.assertNoFiles(slots, REASON)).rejects.toThrow(REASON); - expect(logger.error).toHaveBeenCalledWith( - 'Failed to delete the files of a rejected engine 2.0 payload', - expect.objectContaining({ error: expect.any(Error) }), + expect(() => guard.assertNoFiles(slots, REASON)).toThrow(REASON); + await vi.waitFor(() => + expect(logger.error).toHaveBeenCalledWith( + 'Failed to delete the files of a rejected engine 2.0 payload', + expect.objectContaining({ error: expect.any(Error) }), + ), ); }); }); - - describe('discardFiles', () => { - it('deletes the stored files without refusing', async () => { - const slots = [[{ json: {}, binary: { attachment: stored('filesystem:abc') } }]]; - - await expect(guard.discardFiles(slots)).resolves.toBeUndefined(); - expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledExactlyOnceWith([ - 'filesystem:abc', - ]); - }); - - it('does nothing for a payload with no files', async () => { - await expect(guard.discardFiles([[{ json: {} }]])).resolves.toBeUndefined(); - expect(binaryDataService.deleteManyByBinaryDataId).not.toHaveBeenCalled(); - }); - - it('never throws when the delete fails, so the caller keeps its own answer', async () => { - binaryDataService.deleteManyByBinaryDataId.mockRejectedValue(new Error('store is down')); - const slots = [[{ json: {}, binary: { attachment: stored('s3:abc') } }]]; - - await expect(guard.discardFiles(slots)).resolves.toBeUndefined(); - expect(logger.error).toHaveBeenCalled(); - }); - }); }); diff --git a/packages/cli/src/services/engine-v2-payload-guard.service.ts b/packages/cli/src/services/engine-v2-payload-guard.service.ts index d4b862799b9..93d6a97f7e9 100644 --- a/packages/cli/src/services/engine-v2-payload-guard.service.ts +++ b/packages/cli/src/services/engine-v2-payload-guard.service.ts @@ -32,27 +32,21 @@ export class EngineV2PayloadGuard { * * `reason` names the surface, so the user hears which trigger could not be * served rather than a generic message. + * + * Refuses synchronously, so a caller that cannot await (a poll's `__emit`) + * still refuses before doing anything that only makes sense for a run it is + * about to start. The delete runs detached: nobody reads its result, and + * {@link deleteStoredFiles} never throws. */ - async assertNoFiles(slots: PayloadSlots, reason: string): Promise { + assertNoFiles(slots: PayloadSlots, reason: string): void { const files = this.filesIn(slots); if (files.length === 0) return; - await this.deleteStoredFiles(files); + void this.deleteStoredFiles(files); throw new UserError(reason); } - /** - * Deletes the files of a payload that is being discarded for some other - * reason. Never throws: the caller's own refusal is the answer. - */ - async discardFiles(slots: PayloadSlots): Promise { - const files = this.filesIn(slots); - if (files.length === 0) return; - - await this.deleteStoredFiles(files); - } - /** An empty `binary` map carries no file, so it does not count. */ private filesIn(slots: PayloadSlots): IBinaryData[] { return slots.flatMap((slot) => slot ?? []).flatMap((item) => Object.values(item.binary ?? {})); @@ -60,8 +54,9 @@ export class EngineV2PayloadGuard { /** * Only stored modes give a file an id; in memory the data rides on the item and - * there is nothing to delete. A failed delete leaks a file, which must not - * replace the caller's reason with a storage error. + * there is nothing to delete. Never throws: a failed delete leaks a file, which + * must not replace the caller's reason with a storage error, and callers detach + * this call. */ private async deleteStoredFiles(files: IBinaryData[]): Promise { const storedIds = files.map((file) => file.id).filter((id) => id !== undefined); diff --git a/packages/cli/src/webhooks/engine-v2-webhooks.ts b/packages/cli/src/webhooks/engine-v2-webhooks.ts index 2549b5b0d2a..a3012d36581 100644 --- a/packages/cli/src/webhooks/engine-v2-webhooks.ts +++ b/packages/cli/src/webhooks/engine-v2-webhooks.ts @@ -113,8 +113,8 @@ export class EngineV2Webhooks { * Only the webhook node's own output says whether the request brought a file, * so this runs after the node, unlike {@link assertSupported}. */ - async assertPayloadSupported(webhookResultData: IWebhookResponseData): Promise { - await this.payloadGuard.assertNoFiles( + assertPayloadSupported(webhookResultData: IWebhookResponseData): void { + this.payloadGuard.assertNoFiles( webhookResultData.workflowData ?? [], 'Engine 2.0 cannot receive files from a webhook yet.', ); diff --git a/packages/cli/src/webhooks/webhook-helpers.ts b/packages/cli/src/webhooks/webhook-helpers.ts index 004c6066469..9b7c619c91f 100644 --- a/packages/cli/src/webhooks/webhook-helpers.ts +++ b/packages/cli/src/webhooks/webhook-helpers.ts @@ -916,7 +916,7 @@ export async function executeWebhook( // The node's output is the only place a file shows up, so this cannot run with // the checks above. - if (routesToEngineV2) await engineV2Webhooks.assertPayloadSupported(webhookResultData); + if (routesToEngineV2) engineV2Webhooks.assertPayloadSupported(webhookResultData); // Reactive credential-status gate. Runs only once we know the workflow will // execute (workflowData is defined), so a falsy "Only Run If" short-circuits diff --git a/packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts b/packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts index 384313ccaca..3fce109ade4 100644 --- a/packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts +++ b/packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts @@ -414,6 +414,108 @@ describe('WorkflowExecutionService', () => { }); }); + describe('runPolledWorkflowV2()', () => { + const node = mock({ id: 'node-1', name: 'Poll Node' }); + const workflow = mock({ + id: 'wf-1', + active: true, + activeVersionId: 'some-version-id', + nodes: [node], + }); + const cursor = { lastItemId: 'a' }; + const pollItems = [[{ json: { id: 1 } }]]; + let responsePromise: MockProxy>; + + const runPolledWorkflowV2 = async () => + await workflowExecutionService.runPolledWorkflowV2( + workflow, + node, + pollItems, + additionalData, + 'trigger', + cursor, + responsePromise, + ); + + beforeEach(() => { + vi.clearAllMocks(); + responsePromise = mock>(); + workflowRunner.run.mockResolvedValue('exec-v2'); + pollCursorService.commitCursorOnly.mockResolvedValue(true); + }); + + test('starts the run before committing the cursor, and returns its id', async () => { + const returned = await runPolledWorkflowV2(); + + expect(returned).toBe('exec-v2'); + expect(workflowRunner.run).toHaveBeenCalledWith( + expect.objectContaining({ + workflowData: workflow, + executionData: expect.objectContaining({ + executionData: expect.objectContaining({ + nodeExecutionStack: [{ node, data: { main: pollItems }, source: null }], + }), + }), + }), + true, + undefined, + undefined, + responsePromise, + ); + expect(pollCursorService.commitCursorOnly).toHaveBeenCalledWith({ + workflowId: 'wf-1', + nodeId: 'node-1', + cursor, + fence: undefined, + }); + // The cursor must not advance until the data plane has confirmed the + // run: a commit before that could advance past items nothing carried. + expect(workflowRunner.run.mock.invocationCallOrder[0]).toBeLessThan( + pollCursorService.commitCursorOnly.mock.invocationCallOrder[0], + ); + }); + + test('never commits the cursor when the run fails to start', async () => { + const runError = new Error('engine 2.0 rejected the run'); + workflowRunner.run.mockRejectedValue(runError); + + await expect(runPolledWorkflowV2()).rejects.toBe(runError); + + expect(pollCursorService.commitCursorOnly).not.toHaveBeenCalled(); + }); + + test('logs, but does not fail the run, when the lease is gone by the time it commits', async () => { + pollCursorService.commitCursorOnly.mockResolvedValue(false); + + const returned = await runPolledWorkflowV2(); + + expect(returned).toBe('exec-v2'); + expect(logger.warn).toHaveBeenCalledWith( + 'Poll cursor commit skipped after its execution already started: the poll no longer holds its lease', + { workflowId: 'wf-1', nodeId: 'node-1', nodeName: 'Poll Node', executionId: 'exec-v2' }, + ); + }); + + test('passes the fence through to the cursor commit', async () => { + const fence = { taskId: 'task-1', leaseEpoch: 3 }; + + await workflowExecutionService.runPolledWorkflowV2( + workflow, + node, + pollItems, + additionalData, + 'trigger', + cursor, + responsePromise, + fence, + ); + + expect(pollCursorService.commitCursorOnly).toHaveBeenCalledWith( + expect.objectContaining({ fence }), + ); + }); + }); + describe('executeManually()', () => { beforeEach(() => { workflowRunner.run.mockClear(); diff --git a/packages/cli/src/workflows/triggers/__tests__/engine-v2-active-triggers.test.ts b/packages/cli/src/workflows/triggers/__tests__/engine-v2-active-triggers.test.ts index eef29a82d49..61ac13e04cd 100644 --- a/packages/cli/src/workflows/triggers/__tests__/engine-v2-active-triggers.test.ts +++ b/packages/cli/src/workflows/triggers/__tests__/engine-v2-active-triggers.test.ts @@ -16,8 +16,7 @@ describe('EngineV2ActiveTriggers', () => { beforeEach(() => { vi.clearAllMocks(); - payloadGuard.assertNoFiles.mockResolvedValue(undefined); - payloadGuard.discardFiles.mockResolvedValue(undefined); + payloadGuard.assertNoFiles.mockReturnValue(undefined); engineV2ActiveTriggers = new EngineV2ActiveTriggers(dispatcher, payloadGuard); }); @@ -60,10 +59,10 @@ describe('EngineV2ActiveTriggers', () => { }); describe('assertPayloadSupported', () => { - it('asks the guard to refuse files, naming the trigger surface', async () => { + it('asks the guard to refuse files, naming the trigger surface', () => { const slots = [[{ json: {}, binary: { data: mock() } }]]; - await engineV2ActiveTriggers.assertPayloadSupported(slots); + engineV2ActiveTriggers.assertPayloadSupported(slots); expect(payloadGuard.assertNoFiles).toHaveBeenCalledWith( slots, @@ -71,29 +70,12 @@ describe('EngineV2ActiveTriggers', () => { ); }); - it('surfaces the refusal the guard raises', async () => { - payloadGuard.assertNoFiles.mockRejectedValue(new UserError('nope')); + it('surfaces the refusal the guard raises, synchronously', () => { + payloadGuard.assertNoFiles.mockImplementation(() => { + throw new UserError('nope'); + }); - await expect(engineV2ActiveTriggers.assertPayloadSupported([[]])).rejects.toThrow('nope'); - }); - }); - - describe('discardFiles', () => { - it('hands the payload to the guard', async () => { - const slots = [[{ json: {} }]]; - - await engineV2ActiveTriggers.discardFiles(slots); - - expect(payloadGuard.discardFiles).toHaveBeenCalledWith(slots); - }); - }); - - describe('assertPollSupported', () => { - it('always refuses', () => { - expect(() => engineV2ActiveTriggers.assertPollSupported()).toThrow(UserError); - expect(() => engineV2ActiveTriggers.assertPollSupported()).toThrow( - 'Engine 2.0 cannot run polling triggers yet.', - ); + expect(() => engineV2ActiveTriggers.assertPayloadSupported([[]])).toThrow('nope'); }); }); }); 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 4d674d2db6a..e80cd683bbd 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 @@ -947,6 +947,47 @@ describe('TriggerExecutionContextFactory', () => { expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled(); }); + test('routes to runPolledWorkflowV2, not the transactional runPolledWorkflow, on engine 2.0', async () => { + engineV2Dispatcher.handlesWorkflow.mockReturnValue(true); + workflowExecutionService.runPolledWorkflowV2.mockResolvedValue('exec-v2'); + const responsePromise = createDeferredPromise(); + + await context.__runPoll(async () => { + Object.assign(context.getWorkflowStaticData('node'), { lastItemId: 'a' }); + context.__emit(pollData, responsePromise); + }); + await sleep(0); + + expect(workflowExecutionService.runPolledWorkflowV2).toHaveBeenCalledWith( + expect.objectContaining({ id: 'wf-1' }), + node, + pollData, + additionalData, + mode, + { lastItemId: 'a' }, + responsePromise, + undefined, + ); + expect(workflowExecutionService.runPolledWorkflow).not.toHaveBeenCalled(); + expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled(); + }); + + test('refuses a migrated poll that carries a file, and commits no cursor, on engine 2.0', async () => { + engineV2Dispatcher.handlesWorkflow.mockReturnValue(true); + const storedFile = mock({ id: 'filesystem:abc', mimeType: 'text/plain' }); + const fileData: INodeExecutionData[][] = [[{ json: {}, binary: { attachment: storedFile } }]]; + + await context.__runPoll(async () => { + Object.assign(context.getWorkflowStaticData('node'), { lastItemId: 'a' }); + expect(() => context.__emit(fileData)).toThrow( + 'Engine 2.0 cannot receive files from a trigger yet.', + ); + }); + + expect(pollCursorService.commitCursorOnly).not.toHaveBeenCalled(); + expect(workflowExecutionService.runPolledWorkflowV2).not.toHaveBeenCalled(); + }); + // A poll that neither emitted nor committed leaves its mutation behind; the next // poll must not pick it up and commit an advance past items nobody carried. const abandonedStaging = [ @@ -1823,16 +1864,58 @@ describe('TriggerExecutionContextFactory', () => { ]); }); - test('refuses the emit and commits no cursor', () => { - expect(() => pollContext().__emit([[{ json: {} }]])).toThrow( - 'Engine 2.0 cannot run polling triggers yet.', + test('refuses an emit carrying a file, and commits no cursor', () => { + const data: INodeExecutionData[][] = [[{ json: {}, binary: { attachment: storedFile } }]]; + + expect(() => pollContext().__emit(data)).toThrow( + 'Engine 2.0 cannot receive files from a trigger yet.', ); expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled(); - expect(workflowExecutionService.runPolledWorkflow).not.toHaveBeenCalled(); + expect(workflowExecutionService.runPolledWorkflowV2).not.toHaveBeenCalled(); expect(pollCursorService.commitWithExecution).not.toHaveBeenCalled(); expect(pollCursorService.commitCursorOnly).not.toHaveBeenCalled(); expect(workflowStaticDataService.saveStaticData).not.toHaveBeenCalled(); }); + + test('hands an unmigrated emit to the plain runner, which dispatches it', async () => { + pollContext().__emit([[{ json: {} }]]); + await sleep(0); + + expect(workflowExecutionService.runWorkflow).toHaveBeenCalled(); + expect(workflowExecutionService.runPolledWorkflowV2).not.toHaveBeenCalled(); + }); + + test('re-checks the payload against fresh data when the registration snapshot missed the transition to engine 2.0', async () => { + const staleWorkflowData = mock({ id: 'wf-1', name: 'Test Workflow' }); + const freshWorkflowData = mock({ id: 'wf-1', name: 'Test Workflow' }); + // The registration snapshot still says v1; only the fresh read reflects + // the workflow having since been republished onto engine 2.0. + engineV2Dispatcher.handlesWorkflow.mockImplementation((wf) => wf === freshWorkflowData); + + const getPollFunctions = factory.getExecutePollFunctions( + staleWorkflowData, + additionalData, + mode, + activation, + async () => freshWorkflowData, + ); + const context = getPollFunctions( + workflow, + mock({ name: 'Poll Node' }), + additionalData, + mode, + activation, + ); + + const data: INodeExecutionData[][] = [[{ json: {}, binary: { attachment: storedFile } }]]; + context.__emit(data); + await sleep(0); + + expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled(); + expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledExactlyOnceWith([ + 'filesystem:abc', + ]); + }); }); describe('a workflow that did not opt in', () => { diff --git a/packages/cli/src/workflows/triggers/engine-v2-active-triggers.ts b/packages/cli/src/workflows/triggers/engine-v2-active-triggers.ts index aab62efce7f..33666410b4c 100644 --- a/packages/cli/src/workflows/triggers/engine-v2-active-triggers.ts +++ b/packages/cli/src/workflows/triggers/engine-v2-active-triggers.ts @@ -68,30 +68,11 @@ export class EngineV2ActiveTriggers { * the node's own output says whether it produced one, so this runs on the emit * rather than at activation. Email Read IMAP is the case that matters today: * it downloads attachments and passes no promise, so nothing else refuses it. - */ - async assertPayloadSupported(slots: Array): Promise { - await this.payloadGuard.assertNoFiles( - slots, - 'Engine 2.0 cannot receive files from a trigger yet.', - ); - } - - /** - * Deletes the files of an emit refused for another reason, so a payload the - * poll path turns away does not leave them behind. Never throws. - */ - async discardFiles(slots: Array): Promise { - await this.payloadGuard.discardFiles(slots); - } - - /** - * Rejects a polled emit. * - * A migrated poll node commits its cursor in the same transaction as the - * execution row, which a v2 run does not create. - * TODO(CAT-4078): commit the cursor against a data-plane execution instead. + * Refuses synchronously, so a poll's `__emit` can call it before it takes the + * staged cursor. */ - assertPollSupported(): never { - throw new UserError('Engine 2.0 cannot run polling triggers yet.'); + assertPayloadSupported(slots: Array): void { + this.payloadGuard.assertNoFiles(slots, 'Engine 2.0 cannot receive files from a trigger yet.'); } } 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 c37bd83a71f..d45efda82bf 100644 --- a/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts +++ b/packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts @@ -116,15 +116,15 @@ export class TriggerExecutionContextFactory { * The done promise needs no help here: {@link settleDonePromise} rejects it * from the returned promise chain. */ - private async assertEngineV2Supported( + private assertEngineV2Supported( data: INodeExecutionData[][], emit: EngineV2ActiveTriggerEmit, - ): Promise { + ): void { try { // Files first, because this check deletes what it refuses: a refusal for // any other reason would otherwise leave the stored files behind, owned by // no execution. - await this.engineV2ActiveTriggers.assertPayloadSupported(data); + this.engineV2ActiveTriggers.assertPayloadSupported(data); this.engineV2ActiveTriggers.assertSupported(emit); } catch (error) { emit.responsePromise?.reject(ensureError(error)); @@ -209,7 +209,7 @@ export class TriggerExecutionContextFactory { // Checked against the fresh data, so this agrees with the dispatcher, // which decides on the same copy. if (this.engineV2ActiveTriggers.handles(freshWorkflowData, mode)) { - await this.assertEngineV2Supported(data, { responsePromise, donePromise }); + this.assertEngineV2Supported(data, { responsePromise, donePromise }); } return await this.workflowExecutionService.runWorkflow( @@ -372,12 +372,7 @@ export class TriggerExecutionContextFactory { // retried. Reads the registration's copy of the workflow rather than the // fresh one for the same reason: the fresh read comes too late. if (this.engineV2ActiveTriggers.handles(workflowData, mode)) { - // Detached because `__emit` is synchronous. The poll may have stored - // attachments, and no execution will ever own them. - void this.engineV2ActiveTriggers - .discardFiles(data) - .catch((error: unknown) => this.logTriggerExecutionFailure(error, workflowData, node)); - this.engineV2ActiveTriggers.assertPollSupported(); + this.engineV2ActiveTriggers.assertPayloadSupported(data); } const cursor = takeStagedCursor(); @@ -390,15 +385,42 @@ export class TriggerExecutionContextFactory { // TODO(CAT-3202): resolves workflow data via callback so we // can feature-flag between in-memory data and the published data // service. Once the flag is removed, we'll call the service directly. - const executePromise = resolveWorkflowData().then(async (freshWorkflowData) => - cursor === null - ? await this.workflowExecutionService.runWorkflow( + const executePromise = resolveWorkflowData().then(async (freshWorkflowData) => { + // The registration snapshot above can be stale by the time this + // resolves (e.g. the workflow was just republished onto engine 2.0), + // so a payload that slipped past that check is guarded again here, + // against the copy that actually decides where this run goes. + const routesToV2 = this.engineV2ActiveTriggers.handles(freshWorkflowData, mode); + if (routesToV2) { + try { + this.engineV2ActiveTriggers.assertPayloadSupported(data); + } catch (error) { + responsePromise?.reject(ensureError(error)); + throw error; + } + } + + if (cursor === null) { + return await this.workflowExecutionService.runWorkflow( + freshWorkflowData, + node, + data, + additionalData, + mode, + responsePromise, + ); + } + + return routesToV2 + ? await this.workflowExecutionService.runPolledWorkflowV2( freshWorkflowData, node, data, additionalData, mode, + cursor, responsePromise, + fence, ) : await this.workflowExecutionService.runPolledWorkflow( freshWorkflowData, @@ -409,8 +431,8 @@ export class TriggerExecutionContextFactory { cursor, responsePromise, fence, - ), - ); + ); + }); if (donePromise) this.settleDonePromise(executePromise, donePromise); diff --git a/packages/cli/src/workflows/workflow-execution.service.ts b/packages/cli/src/workflows/workflow-execution.service.ts index 38cde6232b5..d67af73fa58 100644 --- a/packages/cli/src/workflows/workflow-execution.service.ts +++ b/packages/cli/src/workflows/workflow-execution.service.ts @@ -259,6 +259,53 @@ export class WorkflowExecutionService { return executionId; } + /** + * Starts a polled execution on engine 2.0, then advances the cursor. + * + * The v2 path keeps no control-plane execution row, so the cursor cannot + * commit in the same transaction as the run the way {@link runPolledWorkflow} + * does. It commits after the data plane confirms the run started instead: a + * crash between the two calls can duplicate a poll, never lose one. + * TODO(CAT-4078): add a dedup key once `StartExecutionRequest` supports one. + */ + async runPolledWorkflowV2( + workflowData: IWorkflowBase, + node: INode, + data: INodeExecutionData[][], + additionalData: IWorkflowExecuteAdditionalData, + mode: WorkflowExecuteMode, + cursor: PollCursor, + responsePromise?: IDeferredPromise, + fence?: PollLeaseFence, + ): Promise { + const executionId = await this.runWorkflow( + workflowData, + node, + data, + additionalData, + mode, + responsePromise, + ); + + const committed = await this.pollCursorService.commitCursorOnly({ + workflowId: workflowData.id, + nodeId: node.id, + cursor, + fence, + }); + + if (!committed) { + // The run already started; the lease loss only means a concurrent + // poller may repeat the same window, not that anything was lost. + this.logger.warn( + 'Poll cursor commit skipped after its execution already started: the poll no longer holds its lease', + { workflowId: workflowData.id, nodeId: node.id, nodeName: node.name, executionId }, + ); + } + + return executionId; + } + /** * Marks a committed row that failed to start as crashed, so it does not sit at * `new` indefinitely.