fix(workflow-engines): race condition when retry interval is used

.changeset/fifty-pigs-sit.md

@@ -0,0 +1,8 @@
+---
+"@medusajs/workflow-engine-inmemory": patch
+"@medusajs/workflow-engine-redis": patch
+"@medusajs/orchestration": patch
+"@medusajs/workflows-sdk": patch
+---
+
+fix: workflow retry interval race condition

integration-tests/modules/__tests__/workflow-engine/tests.ts

@@ -185,7 +185,7 @@ export const workflowEngineTestSuite = (
                   hasAsyncSteps: true,
                   hasFailedSteps: false,
                   hasSkippedSteps: false,
-                  hasWaitingSteps: false,
+                  hasWaitingSteps: true,
                   hasRevertedSteps: false,
                 }),
                 context: expect.objectContaining({
@@ -236,6 +236,13 @@ export const workflowEngineTestSuite = (
                 workflow_id: "my-workflow-name",
                 transaction_id: "trx_123",
                 state: "done",
+                execution: expect.objectContaining({
+                  hasAsyncSteps: true,
+                  hasFailedSteps: false,
+                  hasSkippedSteps: false,
+                  hasWaitingSteps: false,
+                  hasRevertedSteps: false,
+                }),
                 context: expect.objectContaining({
                   data: expect.objectContaining({
                     invoke: expect.objectContaining({

packages/core/orchestration/src/transaction/distributed-transaction.ts

@@ -2,14 +2,14 @@ import { isDefined } from "@medusajs/utils"
 import { EventEmitter } from "events"
 import { IDistributedTransactionStorage } from "./datastore/abstract-storage"
 import { BaseInMemoryDistributedTransactionStorage } from "./datastore/base-in-memory-storage"
+import { NonSerializableCheckPointError } from "./errors"
 import { TransactionOrchestrator } from "./transaction-orchestrator"
 import { TransactionStep, TransactionStepHandler } from "./transaction-step"
 import {
   TransactionFlow,
   TransactionHandlerType,
   TransactionState,
 } from "./types"
-import { NonSerializableCheckPointError } from "./errors"
 
 /**
  * @typedef TransactionMetadata
@@ -248,7 +248,6 @@ class DistributedTransaction extends EventEmitter {
       return
     }
 
-    await this.saveCheckpoint()
     await DistributedTransaction.keyValueStore.scheduleRetry(
       this,
       step,
@@ -267,7 +266,6 @@ class DistributedTransaction extends EventEmitter {
       return
     }
 
-    await this.saveCheckpoint()
     await DistributedTransaction.keyValueStore.scheduleTransactionTimeout(
       this,
       Date.now(),

packages/core/orchestration/src/transaction/errors.ts

@@ -84,3 +84,12 @@ export class NonSerializableCheckPointError extends Error {
     this.name = "NonSerializableCheckPointError"
   }
 }
+
+export class SkipExecutionError extends Error {
+  static isSkipExecutionError(error: Error): error is SkipExecutionError {
+    return (
+      error instanceof SkipExecutionError ||
+      error?.name === "SkipExecutionError"
+    )
+  }
+}