continue on failure

Author: carlos-r-l-rodrigues

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

@@ -1192,6 +1192,10 @@ export class TransactionOrchestrator extends EventEmitter {
     isPermanent: boolean,
     response?: unknown
   ): Promise<void> {
+    const isAsync = step.isCompensating()
+      ? step.definition.compensateAsync
+      : step.definition.async
+
     if (isDefined(response) && step.saveResponse) {
       transaction.addResponse(
         step.definition.action!,
@@ -1210,10 +1214,15 @@ export class TransactionOrchestrator extends EventEmitter {
     )
 
     if (ret.transactionIsCancelling) {
-      return await this.cancelTransaction(transaction, {
+      await this.cancelTransaction(transaction, {
         preventExecuteNext: true,
       })
     }
+
+    if (isAsync && !ret.stopExecution) {
+      // Schedule to continue the execution of async steps because they are not awaited on purpose and can be handled by another machine
+      await transaction.scheduleRetry(step, 0)
+    }
   }
 
   /**

packages/modules/workflow-engine-inmemory/integration-tests/__fixtures__/index.ts

@@ -3,7 +3,8 @@ export * from "./workflow_2"
 export * from "./workflow_async"
 export * from "./workflow_conditional_step"
 export * from "./workflow_idempotent"
+export * from "./workflow_not_idempotent_with_retention"
+export * from "./workflow_parallel_async"
 export * from "./workflow_step_timeout"
 export * from "./workflow_sync"
 export * from "./workflow_transaction_timeout"
-export * from "./workflow_not_idempotent_with_retention"

packages/modules/workflow-engine-inmemory/integration-tests/__fixtures__/workflow_parallel_async.ts

@@ -0,0 +1,76 @@
+import { Modules } from "@medusajs/framework/utils"
+import {
+  createStep,
+  createWorkflow,
+  parallelize,
+  StepResponse,
+} from "@medusajs/framework/workflows-sdk"
+
+const step_2 = createStep(
+  {
+    name: "step_2",
+    async: true,
+  },
+  async (_, { container }) => {
+    const we = container.resolve(Modules.WORKFLOW_ENGINE)
+
+    await we.run("workflow_sub_workflow", {
+      throwOnError: true,
+    })
+  }
+)
+
+const parallelStep2Invoke = jest.fn(() => {
+  throw new Error("Error in parallel step")
+})
+const step_2_sub = createStep(
+  {
+    name: "step_2",
+    async: true,
+  },
+  parallelStep2Invoke
+)
+
+const subFlow = createWorkflow(
+  {
+    name: "workflow_sub_workflow",
+    retentionTime: 1000,
+  },
+  function (input) {
+    step_2_sub()
+  }
+)
+
+const step_1 = createStep(
+  {
+    name: "step_1",
+    async: true,
+  },
+  jest.fn(() => {
+    return new StepResponse("step_1")
+  })
+)
+
+const parallelStep3Invoke = jest.fn(() => {
+  return new StepResponse({
+    done: true,
+  })
+})
+
+const step_3 = createStep(
+  {
+    name: "step_3",
+    async: true,
+  },
+  parallelStep3Invoke
+)
+
+createWorkflow(
+  {
+    name: "workflow_parallel_async",
+    retentionTime: 1000,
+  },
+  function (input) {
+    parallelize(step_1(), step_2(), step_3())
+  }
+)

packages/modules/workflow-engine-inmemory/integration-tests/__tests__/index.spec.ts

@@ -1,3 +1,4 @@
+import { MedusaContainer } from "@medusajs/framework"
 import {
   DistributedTransactionType,
   TransactionState,
@@ -43,7 +44,6 @@ import {
   workflowEventGroupIdStep2Mock,
 } from "../__fixtures__/workflow_event_group_id"
 import { createScheduled } from "../__fixtures__/workflow_scheduled"
-import { container, MedusaContainer } from "@medusajs/framework"
 
 jest.setTimeout(60000)
 
@@ -912,7 +912,6 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
 
           expect(spy).toHaveBeenCalledTimes(1)
 
-          console.log(spy.mock.results)
           expect(spy).toHaveReturnedWith(
             expect.objectContaining({ output: { testValue: "test" } })
           )
@@ -958,6 +957,35 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
           expect(executionsList).toHaveLength(1)
           expect(executionsListAfter).toHaveLength(1)
         })
+
+        it("should display error when multple async steps are running in parallel", (done) => {
+          void workflowOrcModule.run("workflow_parallel_async", {
+            input: {},
+            throwOnError: false,
+          })
+
+          void workflowOrcModule.subscribe({
+            workflowId: "workflow_parallel_async",
+            subscriber: (event) => {
+              if (event.eventType === "onFinish") {
+                done()
+                expect(event.errors).toEqual(
+                  expect.arrayContaining([
+                    expect.objectContaining({
+                      action: "step_2",
+                      handlerType: "invoke",
+                      error: expect.objectContaining({
+                        message: "Error in parallel step",
+                      }),
+                    }),
+                  ])
+                )
+              }
+            },
+          })
+
+          failTrap(done)
+        })
       })
 
       describe("Cleaner job", function () {