From 9c333bb046ca37e701418820e54a7fc84311317c Mon Sep 17 00:00:00 2001 From: Joan Planas <114454447+joanproductivo@users.noreply.github.com> Date: Mon, 26 Oct 2026 11:21:09 +0100 Subject: [PATCH] fix(workflows): resume a waiting step once handleEvent reads the executions waiting for an event once and then resumes each of them. By the time it reaches one, another delivery of the same event may already have resumed it: two notifications for the same email at once (link scanners clicking every link, repeated clicks), or the same event tracked again while an earlier wait of the loop was resumed. The step was then marked completed a second time and the execution ran its next step again, sending the follow-up email twice. A timeout firing as the event arrived had the same effect, taking both the event and the timeout branch. The wait is now claimed with an update conditional on WAITING, in handleEvent and in processTimeout, and whoever moves it out of WAITING is the one that resumes it. Cancelling the timeout job after a claim is best effort: BullMQ cannot remove a job that is already running, and that job will now find the wait claimed and stop. Co-Authored-By: Claude Opus 5.5 --- .../src/services/WorkflowExecutionService.ts | 33 +++- ...orkflowExecutionService.waitResume.test.ts | 154 ++++++++++++++++++ 2 files changed, 179 insertions(+), 8 deletions(-) create mode 100644 apps/api/src/services/__tests__/WorkflowExecutionService.waitResume.test.ts diff --git a/apps/api/src/services/WorkflowExecutionService.ts b/apps/api/src/services/WorkflowExecutionService.ts index 8afb3b0f7..3da320406 100644 --- a/apps/api/src/services/WorkflowExecutionService.ts +++ b/apps/api/src/services/WorkflowExecutionService.ts @@ -328,9 +328,9 @@ export class WorkflowExecutionService { return; } - // Mark step as completed with timeout - await prisma.workflowStepExecution.update({ - where: {id: stepExecution.id}, + // Mark step as completed with timeout, unless the event claimed the wait since it was read + const {count} = await prisma.workflowStepExecution.updateMany({ + where: {id: stepExecution.id, status: StepExecutionStatus.WAITING}, data: { status: StepExecutionStatus.COMPLETED, completedAt: new Date(), @@ -346,6 +346,10 @@ export class WorkflowExecutionService { }, }); + if (count === 0) { + return; + } + // Continue workflow - find transitions with timeout/fallback logic const transitions = stepExecution.step.outgoingTransitions || []; const fallbackTransition = transitions.find( @@ -438,9 +442,12 @@ export class WorkflowExecutionService { const config = stepExecution.step.config; if (config && typeof config === 'object' && 'eventName' in config && config.eventName === eventName) { - // Event matches, resume execution - await prisma.workflowStepExecution.update({ - where: {id: stepExecution.id}, + // Event matches. Claim the wait before resuming it: the list above can be stale by now, + // as the same event may be delivered twice at once (SES notifications, repeated clicks) + // or tracked again while an earlier wait of this loop was resumed. Whoever moves the + // step out of WAITING resumes it; everyone else leaves it alone. + const {count} = await prisma.workflowStepExecution.updateMany({ + where: {id: stepExecution.id, status: StepExecutionStatus.WAITING}, data: { status: StepExecutionStatus.COMPLETED, completedAt: new Date(), @@ -452,8 +459,18 @@ export class WorkflowExecutionService { }, }); - // Cancel any pending timeout job - await QueueService.cancelWorkflowTimeout(stepExecution.id); + if (count === 0) { + continue; + } + + // Cancel any pending timeout job. Best effort: a timeout job that is already running + // cannot be removed, and it will find the wait claimed and stop. Failing here would + // leave the claimed execution with nobody to move it on. + try { + await QueueService.cancelWorkflowTimeout(stepExecution.id); + } catch (error) { + signale.warn(`[WORKFLOW] Could not cancel the timeout of step execution ${stepExecution.id}:`, error); + } // Continue workflow await this.processNextSteps(stepExecution.execution, stepExecution.step, {eventReceived: true}); diff --git a/apps/api/src/services/__tests__/WorkflowExecutionService.waitResume.test.ts b/apps/api/src/services/__tests__/WorkflowExecutionService.waitResume.test.ts new file mode 100644 index 000000000..fc797b911 --- /dev/null +++ b/apps/api/src/services/__tests__/WorkflowExecutionService.waitResume.test.ts @@ -0,0 +1,154 @@ +import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest'; +import {StepExecutionStatus, WorkflowExecutionStatus, WorkflowStepType} from '@plunk/db'; +import {prisma as servicePrisma} from '../../database/prisma'; +import {QueueService} from '../QueueService'; +import {WorkflowExecutionService} from '../WorkflowExecutionService'; +import {factories, getPrismaClient} from '../../../../../test/helpers'; + +vi.mock('../QueueService', () => ({ + QueueService: { + queueWorkflowStep: vi.fn(async () => ({id: 'mock-job-id'})), + queueEmail: vi.fn(async () => ({id: 'mock-email-job-id'})), + queueWorkflowTimeout: vi.fn(async () => ({id: 'mock-timeout-job-id'})), + cancelWorkflowTimeout: vi.fn(async () => true), + }, +})); + +const EVENT = 'order.shipped'; + +type Run = Awaited>; + +let projectId: string; +let contactId: string; +const prisma = getPrismaClient(); + +/** TRIGGER → WAIT_FOR_EVENT → SEND_EMAIL, started and parked on the wait. */ +async function createRun() { + const workflow = await factories.createWorkflow({projectId}); + const trigger = await prisma.workflowStep.findFirstOrThrow({ + where: {workflowId: workflow.id, type: WorkflowStepType.TRIGGER}, + }); + const wait = await factories.createWorkflowStep({ + workflowId: workflow.id, + type: WorkflowStepType.WAIT_FOR_EVENT, + config: {eventName: EVENT, timeout: 3600}, + }); + const template = await factories.createTemplate({projectId}); + const send = await factories.createWorkflowStep({ + workflowId: workflow.id, + type: WorkflowStepType.SEND_EMAIL, + templateId: template.id, + }); + await prisma.workflowTransition.create({data: {fromStepId: trigger.id, toStepId: wait.id}}); + await prisma.workflowTransition.create({data: {fromStepId: wait.id, toStepId: send.id}}); + + const execution = await prisma.workflowExecution.create({ + data: {workflowId: workflow.id, contactId, status: WorkflowExecutionStatus.RUNNING, currentStepId: trigger.id}, + }); + await WorkflowExecutionService.processStepExecution(execution.id, trigger.id); + + const waitExecution = () => + prisma.workflowStepExecution.findFirstOrThrow({where: {executionId: execution.id, stepId: wait.id}}); + expect((await waitExecution()).status).toBe(StepExecutionStatus.WAITING); + + return { + executionId: execution.id, + sendStepId: send.id, + waitExecution, + timeOut: async () => WorkflowExecutionService.processTimeout(execution.id, wait.id, (await waitExecution()).id), + }; +} + +/** How many times the step after the wait ran, and how many emails the run sent. */ +async function outcome(run: Run) { + const sends = await prisma.workflowStepExecution.count({ + where: {executionId: run.executionId, stepId: run.sendStepId}, + }); + const emails = await prisma.email.count({where: {workflowExecutionId: run.executionId}}); + return {sends, emails}; +} + +const once = {sends: 1, emails: 1}; + +const ship = () => WorkflowExecutionService.handleEvent(projectId, EVENT, contactId); + +/** + * A wait is resumed once, however many deliveries of its event reach it: the list of waits an + * event reads can be stale by the time it gets to each of them. + */ +describe('WorkflowExecutionService - resuming a wait once', () => { + beforeEach(async () => { + const {project} = await factories.createUserWithProject(); + projectId = project.id; + contactId = (await factories.createContact({projectId, subscribed: true})).id; + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('moves on once when the same event arrives twice at the same time', async () => { + for (let round = 0; round < 5; round++) { + const run = await createRun(); + + await Promise.all([ship(), ship()]); + + expect(await outcome(run)).toEqual(once); + } + }); + + it('moves on once when a resumed execution delivers the event again', async () => { + const first = await createRun(); + const second = await createRun(); + + // While the first execution is being resumed, the same event is tracked again (as a step of + // the resumed execution can do) and resumes the second one. + vi.spyOn(QueueService, 'cancelWorkflowTimeout').mockImplementationOnce(async () => { + await ship(); + }); + + await ship(); + + expect(await outcome(first)).toEqual(once); + expect(await outcome(second)).toEqual(once); + }); + + it('takes one way out when the event and the timeout arrive at the same time', async () => { + const run = await createRun(); + + await Promise.all([ship(), run.timeOut()]); + + expect(await outcome(run)).toEqual(once); + }); + + it('leaves a wait alone when its timeout read it before the event claimed it', async () => { + const run = await createRun(); + const waiting = await servicePrisma.workflowStepExecution.findUniqueOrThrow({ + where: {id: (await run.waitExecution()).id}, + include: {execution: true, step: {include: {outgoingTransitions: {include: {toStep: true}}}}}, + }); + + await ship(); + // The timeout job read the wait while it was still WAITING + vi.spyOn(servicePrisma.workflowStepExecution, 'findUnique').mockResolvedValueOnce(waiting); + await run.timeOut(); + + expect(await outcome(run)).toEqual(once); + expect((await run.waitExecution()).output).toMatchObject({eventName: EVENT, receivedAt: expect.any(String)}); + }); + + it('moves on when the timeout job cannot be cancelled', async () => { + const run = await createRun(); + + // BullMQ refuses to remove a job another worker is running. + vi.spyOn(QueueService, 'cancelWorkflowTimeout').mockRejectedValueOnce( + new Error('Job could not be removed because it is locked by another worker'), + ); + + await ship(); + + expect(await outcome(run)).toEqual(once); + const {status} = await prisma.workflowExecution.findUniqueOrThrow({where: {id: run.executionId}}); + expect(status).toBe(WorkflowExecutionStatus.COMPLETED); + }); +});