wip

Author: carlos-r-l-rodrigues

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

@@ -676,6 +676,8 @@ export class TransactionOrchestrator extends EventEmitter {
     let continueExecution = true
 
     while (continueExecution) {
+      console.log("FLOW", transaction.modelId)
+
       if (transaction.hasFinished()) {
         return
       }
@@ -694,7 +696,7 @@ export class TransactionOrchestrator extends EventEmitter {
         continue
       }
 
-      console.log("nextSteps", JSON.stringify(nextSteps, null, 2))
+      console.log("Remaining STEPS", nextSteps.remaining)
 
       if (nextSteps.remaining === 0) {
         if (transaction.hasTimeout()) {
@@ -709,7 +711,6 @@ export class TransactionOrchestrator extends EventEmitter {
 
       let hasSyncSteps = false
       for (const step of nextSteps.next) {
-        console.log("step", step.id)
         const curState = step.getStates()
         const type = step.isCompensating()
           ? TransactionHandlerType.COMPENSATE
@@ -718,8 +719,6 @@ export class TransactionOrchestrator extends EventEmitter {
         step.lastAttempt = Date.now()
         step.attempts++
 
-        console.log("step current state", curState)
-
         if (curState.state === TransactionStepState.NOT_STARTED) {
           if (!step.startedAt) {
             step.startedAt = Date.now()
@@ -874,6 +873,7 @@ export class TransactionOrchestrator extends EventEmitter {
                 )
               })
               .catch(async (error) => {
+                console.log("ON Failure SYNC", error)
                 if (SkipExecutionError.isSkipExecutionError(error)) {
                   continueExecution = false
                   return
@@ -960,6 +960,8 @@ export class TransactionOrchestrator extends EventEmitter {
                   )
                 })
                 .catch(async (error) => {
+                  console.log("ON Failure", error)
+
                   if (SkipExecutionError.isSkipExecutionError(error)) {
                     continueExecution = false
                     return
@@ -978,6 +980,8 @@ export class TransactionOrchestrator extends EventEmitter {
                   }
 
                   await setStepFailure(error, { response })
+
+                  console.log("ON Failure")
                 })
             })
           )

packages/core/workflows-sdk/src/utils/composer/create-step.ts

@@ -334,11 +334,11 @@ function wrapConditionalStep(
  *   createStep,
  *   StepResponse
  * } from "@medusajs/framework/workflows-sdk"
- * 
+ *
  * interface CreateProductInput {
  *   title: string
  * }
- * 
+ *
  * export const createProductStep = createStep(
  *   "createProductStep",
  *   async function (

packages/core/workflows-sdk/src/utils/composer/create-workflow.ts

@@ -47,22 +47,22 @@ global[OrchestrationUtils.SymbolMedusaWorkflowComposerContext] = null
  *   createProductStep,
  *   getProductStep,
  * } from "./steps"
- * 
+ *
  * interface WorkflowInput {
  *  title: string
  * }
- * 
+ *
  * const myWorkflow = createWorkflow(
  *   "my-workflow",
  *   (input: WorkflowInput) => {
  *    // Everything here will be executed and resolved later
  *    // during the execution. Including the data access.
- * 
+ *
  *     const product = createProductStep(input)
  *     return new WorkflowResponse(getProductStep(product.id))
  *   }
  * )
- * 
+ *
  * export async function GET(
  *   req: MedusaRequest,
  *   res: MedusaResponse
@@ -73,7 +73,7 @@ global[OrchestrationUtils.SymbolMedusaWorkflowComposerContext] = null
  *         title: "Shirt"
  *       }
  *     })
- * 
+ *
  *   res.json({
  *     product
  *   })
@@ -207,6 +207,11 @@ export function createWorkflow<TData, TResult, THooks extends any[]>(
         )
       },
       async (transaction, stepContext) => {
+        console.log(
+          "BEFORE SUB WORKFLOW COMPENSATION",
+          stepContext.idempotencyKey
+        )
+
         if (!transaction) {
           return
         }
@@ -222,6 +227,10 @@ export function createWorkflow<TData, TResult, THooks extends any[]>(
             parentStepIdempotencyKey: stepContext.idempotencyKey,
           },
         })
+
+        console.log("RESPONSE FROM SUB WORKFLOW COMPENSATION", {
+          transaction,
+        })
       }
     )(input) as ReturnType<StepFunction<TData, TResult>>
 

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

@@ -12,10 +12,17 @@ import {
   Modules,
   TransactionHandlerType,
 } from "@medusajs/framework/utils"
+import {
+  createStep,
+  createWorkflow,
+  StepResponse,
+  transform,
+  WorkflowResponse,
+} from "@medusajs/framework/workflows-sdk"
 import { moduleIntegrationTestRunner } from "@medusajs/test-utils"
 import { WorkflowsModuleService } from "@services"
 import { asFunction } from "awilix"
-import { setTimeout as setTimeoutPromise } from "timers/promises"
+import { setTimeout, setTimeout as setTimeoutPromise } from "timers/promises"
 import "../__fixtures__"
 import {
   conditionalStep2Invoke,
@@ -29,14 +36,6 @@ import {
   workflowEventGroupIdStep2Mock,
 } from "../__fixtures__/workflow_event_group_id"
 import { createScheduled } from "../__fixtures__/workflow_scheduled"
-import {
-  createStep,
-  createWorkflow,
-  StepResponse,
-  transform,
-  WorkflowResponse,
-} from "@medusajs/framework/workflows-sdk"
-import { setTimeout } from "timers/promises"
 
 jest.setTimeout(3000000)
 
@@ -126,8 +125,9 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
       const transformMock = jest.fn()
 
       const step0 = createStep(
-        "step0",
+        "RACE_step0",
         async (_) => {
+          console.log("step0")
           step0InvokeMock()
           return new StepResponse("result from step 0")
         },
@@ -137,8 +137,9 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
       )
 
       const step1 = createStep(
-        "step1",
+        "RACE_step1",
         async (_) => {
+          console.log("step1")
           step1InvokeMock()
           await setTimeout(2000)
           throw new Error("error from step 1")
@@ -148,17 +149,18 @@ moduleIntegrationTestRunner<IWorkflowEngineService>({
         }
       )
 
-      const step2 = createStep("step2", async (input: any) => {
+      const step2 = createStep("RACE_step2", async (input: any) => {
+        console.log("step2")
         step2InvokeMock()
         return new StepResponse({ result: input })
       })
 
-      const subWorkflow = createWorkflow("sub-workflow-1", function () {
+      const subWorkflow = createWorkflow("RACE_sub-workflow-1", function () {
         const status = step1()
         return new WorkflowResponse(status)
       })
 
-      createWorkflow("workflow-1", function () {
+      createWorkflow("RACE_workflow-1", function () {
         const build = step0()
 
         const status = subWorkflow.runAsStep({} as any).config({

packages/modules/workflow-engine-redis/src/services/workflow-orchestrator.ts

@@ -189,6 +189,8 @@ export class WorkflowOrchestratorService {
     options?: WorkflowOrchestratorRunOptions<T>,
     @MedusaContext() sharedContext: Context = {}
   ) {
+    console.log("RUN", workflowIdOrWorkflow)
+
     const {
       input,
       transactionId,
@@ -226,6 +228,7 @@ export class WorkflowOrchestratorService {
     const originalOnFinishHandler = events.onFinish!
     delete events.onFinish
 
+    console.log(" Before original RUN")
     const ret = await exportedWorkflow.run({
       input,
       throwOnError: false,
@@ -239,6 +242,7 @@ export class WorkflowOrchestratorService {
     const hasFinished = ret.transaction.hasFinished()
     const metadata = ret.transaction.getFlow().metadata
     const { parentStepIdempotencyKey } = metadata ?? {}
+
     const hasFailed = [
       TransactionState.REVERTED,
       TransactionState.FAILED,
@@ -252,6 +256,8 @@ export class WorkflowOrchestratorService {
       hasFailed,
     }
 
+    console.log({ acknowledgement })
+
     if (hasFinished) {
       const { result, errors } = ret