feat(orchestration): Provide hint in workflows error (#9400)

* feat(orchestration): Provide hint in workflows error

* remove log

* fix tests

* improve stack

* fix type

* formatting

---------

Co-authored-by: Riqwan Thamir <rmthamir@gmail.com>
This commit is contained in:
Adrien de PerettiandRiqwan Thamir authored and GitHub committed 2024-10-02 11:54:07 +02:00
1 parent 7ce9121095
commit 02629625ec
5 files changed
+42 -14

No files matched your search

@@ -549,6 +549,16 @@ export class TransactionOrchestrator extends EventEmitter {
step.changeState(TransactionStepState.TIMEOUT) step.changeState(TransactionStepState.TIMEOUT)
if (error?.stack) {
const workflowId = transaction.modelId
const stepAction = step.definition.action
const sourcePath = transaction.getFlow().metadata?.sourcePath
const sourceStack = sourcePath
? `\n⮑ \sat ${sourcePath}: [${workflowId} -> ${stepAction} (${TransactionHandlerType.INVOKE})]`
: `\n⮑ \sat [${workflowId} -> ${stepAction} (${TransactionHandlerType.INVOKE})]`
error.stack += sourceStack
}
transaction.addError( transaction.addError(
step.definition.action!, step.definition.action!,
TransactionHandlerType.INVOKE, TransactionHandlerType.INVOKE,
@@ -602,13 +612,21 @@ export class TransactionOrchestrator extends EventEmitter {
step.changeStatus(TransactionStepStatus.PERMANENT_FAILURE) step.changeStatus(TransactionStepStatus.PERMANENT_FAILURE)
if (!isTimeout) { if (!isTimeout) {
transaction.addError( const handlerType = step.isCompensating()
step.definition.action!, ? TransactionHandlerType.COMPENSATE
step.isCompensating() : TransactionHandlerType.INVOKE
? TransactionHandlerType.COMPENSATE
: TransactionHandlerType.INVOKE, if (error?.stack) {
error const workflowId = transaction.modelId
) const stepAction = step.definition.action
const sourcePath = transaction.getFlow().metadata?.sourcePath
const sourceStack = sourcePath
? `\n⮑ \sat ${sourcePath}: [${workflowId} -> ${stepAction} (${TransactionHandlerType.INVOKE})]`
: `\n⮑ \sat [${workflowId} -> ${stepAction} (${TransactionHandlerType.INVOKE})]`
error.stack += sourceStack
}
transaction.addError(step.definition.action!, handlerType, error)
} }
if (!step.isCompensating()) { if (!step.isCompensating()) {
@@ -243,6 +243,7 @@ export type TransactionFlow = {
metadata?: { metadata?: {
eventGroupId?: string eventGroupId?: string
parentIdempotencyKey?: string parentIdempotencyKey?: string
sourcePath?: string
[key: string]: unknown [key: string]: unknown
} }
hasAsyncSteps: boolean hasAsyncSteps: boolean
@@ -44,6 +44,7 @@ function createContextualWorkflowRunner<
dataPreparation?: (data: TData) => Promise<unknown> dataPreparation?: (data: TData) => Promise<unknown>
options?: { options?: {
wrappedInput?: boolean wrappedInput?: boolean
sourcePath?: string
} }
container?: LoadedModule[] | MedusaContainer container?: LoadedModule[] | MedusaContainer
}): Omit< }): Omit<
@@ -93,6 +94,7 @@ function createContextualWorkflowRunner<
const flowMetadata = { const flowMetadata = {
eventGroupId, eventGroupId,
parentStepIdempotencyKey, parentStepIdempotencyKey,
sourcePath: options?.sourcePath,
} }
const args = [ const args = [
@@ -334,6 +336,7 @@ export const exportWorkflow = <TData = unknown, TResult = unknown>(
dataPreparation?: (data: TData) => Promise<unknown>, dataPreparation?: (data: TData) => Promise<unknown>,
options?: { options?: {
wrappedInput?: boolean wrappedInput?: boolean
sourcePath?: string
} }
): MainExportedWorkflow<TData, TResult> => { ): MainExportedWorkflow<TData, TResult> => {
function exportedWorkflow< function exportedWorkflow<
@@ -6,19 +6,19 @@ import {
} from "@medusajs/orchestration" } from "@medusajs/orchestration"
import { IEventBusModuleService } from "@medusajs/types" import { IEventBusModuleService } from "@medusajs/types"
import { import {
Modules,
composeMessage, composeMessage,
createMedusaContainer, createMedusaContainer,
Modules,
promiseAll, promiseAll,
} from "@medusajs/utils" } from "@medusajs/utils"
import { asValue } from "awilix" import { asValue } from "awilix"
import { import {
StepResponse,
WorkflowResponse,
createStep, createStep,
createWorkflow, createWorkflow,
parallelize, parallelize,
StepResponse,
transform, transform,
WorkflowResponse,
} from ".." } from ".."
import { MedusaWorkflow } from "../../../medusa-workflow" import { MedusaWorkflow } from "../../../medusa-workflow"
import { createHook } from "../create-hook" import { createHook } from "../create-hook"
@@ -1903,7 +1903,7 @@ describe("Workflow composer", function () {
action: "step1", action: "step1",
handlerType: "invoke", handlerType: "invoke",
error: expect.objectContaining({ error: expect.objectContaining({
message: "invoke fail", message: "invoke fail",
}), }),
}) })
@@ -4,7 +4,11 @@ import {
WorkflowManager, WorkflowManager,
} from "@medusajs/orchestration" } from "@medusajs/orchestration"
import { LoadedModule, MedusaContainer } from "@medusajs/types" import { LoadedModule, MedusaContainer } from "@medusajs/types"
import { OrchestrationUtils, isString } from "@medusajs/utils" import {
getCallerFilePath,
isString,
OrchestrationUtils,
} from "@medusajs/utils"
import { ulid } from "ulid" import { ulid } from "ulid"
import { exportWorkflow } from "../../helper" import { exportWorkflow } from "../../helper"
import { createStep } from "./create-step" import { createStep } from "./create-step"
@@ -34,7 +38,7 @@ global[OrchestrationUtils.SymbolMedusaWorkflowComposerContext] = null
* @returns The created workflow. You can later execute the workflow by invoking it, then using its `run` method. * @returns The created workflow. You can later execute the workflow by invoking it, then using its `run` method.
* *
* @example * @example
* import { * import {
* createWorkflow, * createWorkflow,
* WorkflowResponse * WorkflowResponse
* } from "@medusajs/framework/workflows-sdk" * } from "@medusajs/framework/workflows-sdk"
@@ -50,7 +54,7 @@ global[OrchestrationUtils.SymbolMedusaWorkflowComposerContext] = null
* } * }
* *
* const myWorkflow = createWorkflow( * const myWorkflow = createWorkflow(
* "my-workflow", * "my-workflow",
* (input: WorkflowInput) => { * (input: WorkflowInput) => {
* // Everything here will be executed and resolved later * // Everything here will be executed and resolved later
* // during the execution. Including the data access. * // during the execution. Including the data access.
@@ -92,6 +96,7 @@ export function createWorkflow<TData, TResult, THooks extends any[]>(
input: WorkflowData<TData> input: WorkflowData<TData>
) => void | WorkflowResponse<TResult, THooks> ) => void | WorkflowResponse<TResult, THooks>
): ReturnWorkflow<TData, TResult, THooks> { ): ReturnWorkflow<TData, TResult, THooks> {
const fileSourcePath = getCallerFilePath() as string
const name = isString(nameOrConfig) ? nameOrConfig : nameOrConfig.name const name = isString(nameOrConfig) ? nameOrConfig : nameOrConfig.name
const options = isString(nameOrConfig) ? {} : nameOrConfig const options = isString(nameOrConfig) ? {} : nameOrConfig
@@ -153,6 +158,7 @@ export function createWorkflow<TData, TResult, THooks extends any[]>(
undefined, undefined,
{ {
wrappedInput: true, wrappedInput: true,
sourcePath: fileSourcePath,
} }
) )