fix(workflows-sdk): Miss match context usage within run as step (#12449)

**What**

Currently, runAsStep keep reference of the workflow context that is being run as step, except that the step is composed for the current workflow composition and not the workflow being run as a step. Therefore, the context are currently miss matched leading to wrong configuration being used in case of async workflows.

**BUG**
This fix allow the runAsStep to use the current composition context to configure the step for the sub workflow to be run

**BUG BREAKING**
fix the step config wrongly used to wrap async step handlers. Now steps configured async through .config that returns a new step response will indeed marked itself as success without the need for background execution or calling setStepSuccess (as it was expected originally)

**FEATURE**
This pr also add support for cancelling running transaction, the transaction will be marked as being cancelled, once the current step finished, it will cancel the transaction to start compensating all previous steps including itself

Co-authored-by: Carlos R. L. Rodrigues <37986729+carlos-r-l-rodrigues@users.noreply.github.com>

packages/modules/workflow-engine-redis/integration-tests/__fixtures__/workflow_1.ts

@@ -1,3 +1,4 @@
+import { isPresent } from "@medusajs/framework/utils"
 import {
   createStep,
   createWorkflow,
@@ -25,7 +26,7 @@ const step_1 = createStep(
 const step_2 = createStep(
   "step_2",
   jest.fn((input, context) => {
-    if (input) {
+    if (isPresent(input)) {
       return new StepResponse({ notAsyncResponse: input.hey })
     }
   }),
@@ -54,7 +55,7 @@ createWorkflow("workflow_1", function (input) {
 
   const ret2 = step_2({ hey: "oh" })
 
-  step_2({ hey: "async hello" }).config({
+  step_2().config({
     name: "new_step_name",
     async: true,
   })

packages/modules/workflow-engine-redis/integration-tests/__fixtures__/workflow_2.ts

@@ -1,3 +1,4 @@
+import { isPresent } from "@medusajs/framework/utils"
 import {
   createStep,
   createWorkflow,
@@ -26,7 +27,7 @@ const step_1 = createStep(
 const step_2 = createStep(
   "step_2",
   jest.fn((input, context) => {
-    if (input) {
+    if (isPresent(input)) {
       return new StepResponse({ notAsyncResponse: input.hey })
     }
   }),
@@ -68,7 +69,7 @@ createWorkflow(
 
     const ret2 = step_2({ hey: "oh" })
 
-    step_2({ hey: "async hello" }).config({
+    step_2().config({
       name: "new_step_name",
       async: true,
     })

packages/modules/workflow-engine-redis/integration-tests/__fixtures__/workflow_not_idempotent_with_retention.ts

@@ -3,6 +3,7 @@ import {
   createWorkflow,
   StepResponse,
 } from "@medusajs/framework/workflows-sdk"
+import { isPresent } from "@medusajs/framework/utils"
 
 const step_1 = createStep(
   "step_1",
@@ -23,7 +24,7 @@ const step_1 = createStep(
 
 export const workflowNotIdempotentWithRetentionStep2Invoke = jest.fn(
   (input, context) => {
-    if (input) {
+    if (isPresent(input)) {
       return new StepResponse({ notAsyncResponse: input.hey })
     }
   }

packages/modules/workflow-engine-redis/integration-tests/__fixtures__/workflow_sync.ts

@@ -4,6 +4,7 @@ import {
   StepResponse,
   WorkflowResponse,
 } from "@medusajs/framework/workflows-sdk"
+import { isPresent } from "@medusajs/framework/utils"
 
 const step_1 = createStep(
   "step_1",
@@ -25,7 +26,7 @@ const step_1 = createStep(
 const step_2 = createStep(
   "step_2",
   jest.fn((input, context) => {
-    if (input) {
+    if (isPresent(input)) {
       return new StepResponse({ notAsyncResponse: input.hey })
     }
   }),

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

@@ -30,15 +30,15 @@ import { moduleIntegrationTestRunner } from "@medusajs/test-utils"
 import { asValue } from "awilix"
 import { setTimeout as setTimeoutSync } from "timers"
 import { setTimeout } from "timers/promises"
+import { ulid } from "ulid"
 import { WorkflowsModuleService } from "../../src/services"
 import "../__fixtures__"
-import { createScheduled } from "../__fixtures__/workflow_scheduled"
-import { TestDatabase } from "../utils"
 import {
   workflowNotIdempotentWithRetentionStep2Invoke,
   workflowNotIdempotentWithRetentionStep3Invoke,
 } from "../__fixtures__"
-import { ulid } from "ulid"
+import { createScheduled } from "../__fixtures__/workflow_scheduled"
+import { TestDatabase } from "../utils"
 
 jest.setTimeout(300000)
 
@@ -150,9 +150,220 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
       })
 
       describe("Testing basic workflow", function () {
+        describe("Cancel transaction", function () {
+          it("should cancel an ongoing execution with async unfinished yet step", async () => {
+            const transactionId = "transaction-to-cancel-id"
+            const step1 = createStep("step1", async () => {
+              return new StepResponse("step1")
+            })
+
+            const step2 = createStep("step2", async () => {
+              await setTimeout(500)
+              return new StepResponse("step2")
+            })
+
+            const step3 = createStep("step3", async () => {
+              return new StepResponse("step3")
+            })
+
+            const workflowId = "workflow-to-cancel-id" + ulid()
+
+            createWorkflow(
+              { name: workflowId, retentionTime: 60 },
+              function () {
+                step1()
+                step2().config({ async: true })
+                step3()
+
+                return new WorkflowResponse("finished")
+              }
+            )
+
+            await workflowOrcModule.run(workflowId, {
+              input: {},
+              transactionId,
+            })
+
+            await setTimeout(100)
+
+            await workflowOrcModule.cancel(workflowId, {
+              transactionId,
+            })
+
+            await setTimeout(1000)
+
+            const execution = await workflowOrcModule.listWorkflowExecutions({
+              transaction_id: transactionId,
+            })
+
+            expect(execution.length).toEqual(1)
+            expect(execution[0].state).toEqual(TransactionState.REVERTED)
+          })
+
+          it("should cancel a complete execution with a sync workflow running as async", async () => {
+            const workflowId = "workflow-to-cancel-id" + ulid()
+            const transactionId = "transaction-to-cancel-id"
+            const step1 = createStep("step1", async () => {
+              return new StepResponse("step1")
+            })
+
+            const step2 = createStep("step2", async () => {
+              return new StepResponse("step2")
+            })
+
+            const step3 = createStep("step3", async () => {
+              return new StepResponse("step3")
+            })
+
+            const subWorkflowId = "sub-workflow-id" + ulid()
+            const subWorkflow = createWorkflow(
+              { name: subWorkflowId, retentionTime: 60 },
+              function () {
+                return new WorkflowResponse(step2())
+              }
+            )
+
+            createWorkflow(
+              { name: workflowId, retentionTime: 60 },
+              function () {
+                step1()
+                subWorkflow.runAsStep({ input: {} }).config({ async: true })
+                step3()
+
+                return new WorkflowResponse("finished")
+              }
+            )
+
+            await workflowOrcModule.run(workflowId, {
+              input: {},
+              transactionId,
+            })
+
+            await setTimeout(100)
+
+            await workflowOrcModule.cancel(workflowId, {
+              transactionId,
+            })
+
+            await setTimeout(500)
+
+            const execution = await workflowOrcModule.listWorkflowExecutions({
+              transaction_id: transactionId,
+            })
+
+            expect(execution.length).toEqual(1)
+            expect(execution[0].state).toEqual(TransactionState.REVERTED)
+          })
+
+          it("should cancel an ongoing execution with a sync workflow running as async", async () => {
+            const workflowId = "workflow-to-cancel-id" + ulid()
+            const transactionId = "transaction-to-cancel-id"
+            const step1 = createStep("step1", async () => {
+              return new StepResponse("step1")
+            })
+
+            const step2 = createStep("step2", async () => {
+              await setTimeout(500)
+              return new StepResponse("step2")
+            })
+
+            const step3 = createStep("step3", async () => {
+              return new StepResponse("step3")
+            })
+
+            const subWorkflowId = "sub-workflow-id" + ulid()
+            const subWorkflow = createWorkflow(
+              { name: subWorkflowId, retentionTime: 60 },
+              function () {
+                return new WorkflowResponse(step2())
+              }
+            )
+
+            createWorkflow(
+              { name: workflowId, retentionTime: 60 },
+              function () {
+                step1()
+                subWorkflow.runAsStep({ input: {} }).config({ async: true })
+                step3()
+
+                return new WorkflowResponse("finished")
+              }
+            )
+
+            await workflowOrcModule.run(workflowId, {
+              input: {},
+              transactionId,
+            })
+
+            await setTimeout(100)
+
+            await workflowOrcModule.cancel(workflowId, {
+              transactionId,
+            })
+
+            await setTimeout(1000)
+
+            const execution = await workflowOrcModule.listWorkflowExecutions({
+              transaction_id: transactionId,
+            })
+
+            expect(execution.length).toEqual(1)
+            expect(execution[0].state).toEqual(TransactionState.REVERTED)
+          })
+
+          it("should cancel an ongoing execution with sync steps only", async () => {
+            const transactionId = "transaction-to-cancel-id"
+            const step1 = createStep("step1", async () => {
+              return new StepResponse("step1")
+            })
+
+            const step2 = createStep("step2", async () => {
+              await setTimeout(500)
+              return new StepResponse("step2")
+            })
+
+            const step3 = createStep("step3", async () => {
+              return new StepResponse("step3")
+            })
+
+            const workflowId = "workflow-to-cancel-id" + ulid()
+
+            createWorkflow(
+              { name: workflowId, retentionTime: 60 },
+              function () {
+                step1()
+                step2()
+                step3()
+
+                return new WorkflowResponse("finished")
+              }
+            )
+
+            await workflowOrcModule.run(workflowId, {
+              input: {},
+              transactionId,
+            })
+
+            await setTimeout(100)
+
+            await workflowOrcModule.cancel(workflowId, {
+              transactionId,
+            })
+
+            await setTimeout(1000)
+
+            const execution = await workflowOrcModule.listWorkflowExecutions({
+              transaction_id: transactionId,
+            })
+
+            expect(execution.length).toEqual(1)
+            expect(execution[0].state).toEqual(TransactionState.REVERTED)
+          })
+        })
+
         it("should prevent executing twice the same workflow in perfect concurrency with the same transactionId and non idempotent and not async but retention time is set", async () => {
-          const transactionId = "transaction_id"
-          const workflowId = "workflow_id" + ulid()
+          const transactionId = "concurrency_transaction_id"
+          const workflowId = "concurrency_workflow_id" + ulid()
 
           const step1 = createStep("step1", async () => {
             await setTimeout(100)
@@ -170,10 +381,12 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
           )
 
           const [result1, result2] = await promiseAll([
-            workflowOrcModule.run(workflowId, {
-              input: {},
-              transactionId,
-            }),
+            workflowOrcModule
+              .run(workflowId, {
+                input: {},
+                transactionId,
+              })
+              .catch((e) => e),
             workflowOrcModule
               .run(workflowId, {
                 input: {},
@@ -182,8 +395,8 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
               .catch((e) => e),
           ])
 
-          expect(result1.result).toEqual("step1")
-          expect(result2.message).toEqual(
+          expect(result1.result || result2.result).toEqual("step1")
+          expect(result2.message || result1.message).toEqual(
             "Transaction already started for transactionId: " + transactionId
           )
         })

packages/modules/workflow-engine-redis/src/utils/workflow-orchestrator-storage.ts

@@ -4,6 +4,7 @@ import {
   IDistributedSchedulerStorage,
   IDistributedTransactionStorage,
   SchedulerOptions,
+  SkipCancelledExecutionError,
   SkipExecutionError,
   TransactionCheckpoint,
   TransactionContext,
@@ -632,6 +633,19 @@ export class RedisDistributedTransactionStorage
       throw new SkipExecutionError("Already finished by another execution")
     }
 
+    // First ensure that the latest execution was not cancelled, otherwise we skip the execution
+    const latestTransactionCancelledAt = latestUpdatedFlow.cancelledAt
+    const currentTransactionCancelledAt = currentFlow.cancelledAt
+
+    if (
+      !!latestTransactionCancelledAt &&
+      currentTransactionCancelledAt == null
+    ) {
+      throw new SkipCancelledExecutionError(
+        "Workflow execution has been cancelled during the execution"
+      )
+    }
+
     const currentFlowSteps = Object.values(currentFlow.steps || {})
     const latestUpdatedFlowSteps = latestUpdatedFlow.steps
       ? Object.values(