fix(workflows): resume a waiting step once

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});

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<ReturnType<typeof createRun>>;
+
+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);
+  });
+});