Long-Running Workflows

In this chapter, you'll learn what a long-running workflow is and how to configure it.

What is a Long-Running Workflow?#

When you execute a workflow, you wait until the workflow finishes execution to receive the output.

A long-running workflow is a workflow that continues its execution in the background. You don't receive its output immediately. Instead, you subscribe to the workflow execution to listen to status changes and receive its result once the execution is finished.

Why use Long-Running Workflows?#

Long-running workflows are useful if:

• A task takes too long. For example, you're importing data from a CSV file.
• The workflow's steps wait for an external action to finish before resuming execution. For example, before you import the data from the CSV file, you wait until the import is confirmed by the user.

Configure Long-Running Workflows#

A workflow is considered long-running if at least one step has its Async option set to true and its function returns an empty response.

For example, consider the following workflow and steps:

// step1 is a regular step.
var step1 = composer.CreateStep("step-1",
	func(ctx composer.StepExecutionContext, _ any) (composer.StepResponse[any], error) {
		return composer.NewStepResponse[any](map[string]any{}), nil
	}, nil)

// step2 is an async step: its handler returns an empty response, so the
// step waits for its status to be set from outside.
var step2 = composer.CreateStep("step-2",
	func(ctx composer.StepExecutionContext, _ any) (composer.StepResponse[any], error) {
		slog.Info("Waiting to be successful...")

		return composer.StepResponse[any]{}, nil
	}, nil).Config("", composer.StepOptions{Async: true})

// step3 runs once step2 succeeds.
var step3 = composer.CreateStep("step-3",
	func(ctx composer.StepExecutionContext, _ any) (composer.StepResponse[string], error) {
		return composer.NewStepResponse("Finished three steps"), nil
	}, nil)

The second step has the Async option set to true, and its function returns an empty response (composer.StepResponse[any]{}). This indicates that this step is an asynchronous step. Its type parameter is any, so the output of an empty response is nil.

Then, add the workflow:

// WorkflowOutput is what the workflow returns.
type WorkflowOutput struct {
	Message string `json:"message"`
}

// MyWorkflow is a long-running workflow, since step2 is async and
// returns no response.
var MyWorkflow = composer.CreateWorkflow("hello-world",
	func(e *composer.Execution, _ any) (WorkflowOutput, error) {
		if _, err := step1.Run(e, nil); err != nil {
			return WorkflowOutput{}, err
		}
		if _, err := step2.Run(e, nil); err != nil {
			return WorkflowOutput{}, err
		}
		message, err := step3.Run(e, nil)
		if err != nil {
			return WorkflowOutput{}, err
		}

		return WorkflowOutput{Message: message}, nil
	})

So, when you execute the hello-world workflow, it continues its execution in the background once it reaches the second step. Run returns to its caller when the workflow reaches the async step, and the output it returns is empty: the workflow's output is only available when the workflow finishes.

When is a Workflow Considered Long-Running?#

A workflow is also considered long-running if:

• One of its steps has its Async and BackgroundExecution options set to true, as explained in the next section.
• One of its nested workflows is a long-running workflow.

Since a long-running workflow's execution is stored in the database while it runs, you can view it in the Medusa Admin, as explained in the Store Workflow Executions chapter.

Run a Step in the Background#

You can also make a step run in the background, without waiting for an external action to set its status. Set the BackgroundExecution option of an async step to true. The workflow's execution returns to its caller as soon as the step starts, and the step's function continues in the background. Unlike a step that only waits, its function's response completes the step:

// importStep takes too long to wait for. With BackgroundExecution, the
// workflow's run returns to its caller as soon as the step starts, and the
// step's handler continues in the background. Its response completes the
// step, unlike the response of a step that is only async.
var importStep = composer.CreateStep("import-step",
	func(ctx composer.StepExecutionContext, _ any) (composer.StepResponse[ImportOutput], error) {
		// ... import the data, which takes long
		time.Sleep(10 * time.Millisecond)

		return composer.NewStepResponse(ImportOutput{Imported: 1000}), nil
	}, nil).Config("", composer.StepOptions{Async: true, BackgroundExecution: true})

// ImportWorkflow imports data in the background.
var ImportWorkflow = composer.CreateWorkflow("import-data",
	func(e *composer.Execution, _ any) (ImportOutput, error) {
		return importStep.Run(e, nil)
	})

Change Step Status#

Once the workflow's execution reaches an async step that waits for its status, it waits in the background for the step to succeed or fail before it moves to the next step.

To fail or succeed a step, use composer.SetStepSuccess and composer.SetStepFailure. They find the workflow engine in the container. They're the same operations the Workflow Engine Module's service, registered in the container under the workflows key, has.

Retrieve Transaction ID#

Before changing the status of a workflow execution's async step, you must have the execution's transaction ID.

Execute the workflow with RunWithInfo. The composer.RunInfo value it returns has the details of the execution's transaction. Use its TransactionID to later change async steps' statuses:

// RunAndGetTransactionID executes the workflow and returns the ID of its
// transaction.
func RunAndGetTransactionID(ctx context.Context, container composer.Container) (string, error) {
	_, info, err := MyWorkflow.RunWithInfo(ctx, container, nil)
	if err != nil {
		return "", err
	}

	// use info.TransactionID later to change the status of the async steps
	return info.TransactionID, nil
}

You can also pass your own transaction ID in the options of Run or RunWithInfo, as explained in the Store Workflow Executions chapter.

Change Step Status to Successful#

composer.SetStepSuccess sets a step's status to successful. If you use it on a workflow execution's async step, the workflow continues execution to the next step.

For example:

// SetStep2Success sets the status of step2 of the workflow's execution
// to successful. The workflow then continues with step3.
func SetStep2Success(ctx context.Context, container composer.Container, transactionID string) error {
	return composer.SetStepSuccess(ctx, container, composer.IdempotencyKey{
		Action:        "invoke",
		TransactionID: transactionID,
		StepID:        "step-2",
		WorkflowID:    "hello-world",
	}, "Done!")
}

In this function, you set step-2 of the previous workflow as successful.

SetStepSuccess accepts the following parameters:

Change Step Status to Failed#

composer.SetStepFailure changes a step's status to failed. It accepts the same parameters as SetStepSuccess, but the last one is the error that fails the step.

After changing the async step's status to failed, the workflow execution fails and the compensation functions of previous steps are executed.

For example:

// SetStep2Failure sets the status of step2 to failed. The workflow's
// execution fails, and the compensation functions of the steps that ran
// are executed.
func SetStep2Failure(ctx context.Context, container composer.Container, transactionID string) error {
	return composer.SetStepFailure(ctx, container, composer.IdempotencyKey{
		Action:        "invoke",
		TransactionID: transactionID,
		StepID:        "step-2",
		WorkflowID:    "hello-world",
	}, errors.New("Failed!"))
}

You use this function in a different part of your application, such as an API route, that changes the status of an async step in a long-running workflow's execution to failed.

If no step with the given IDs is waiting for its status, both functions return an error.

Access Long-Running Workflow Status and Result#

To access the status and result of a long-running workflow execution, use composer.Subscribe and composer.Unsubscribe, which are also methods of the Workflow Engine Module's service.

The workflow's execution only notifies its subscribers if you execute it with the Notify option of composer.RunOptions.

For example:

// Start runs the workflow and listens to the changes in the status of its
// execution. It returns the transaction ID of the execution, which you
// use to set the status of its async steps.
func Start(ctx context.Context, container composer.Container, done func(result any)) (string, error) {
	// Notify delivers the events of the run to the subscribers
	_, info, err := MyWorkflow.RunWithInfo(ctx, container, nil, composer.RunOptions{Notify: true})
	if err != nil {
		return "", err
	}

	const subscriberID = "hello-world-subscriber"
	composer.Subscribe("hello-world", info.TransactionID, subscriberID, func(event composer.WorkflowEvent) {
		switch event.EventType {
		case "onFinish":
			// the result is the output of the workflow as a JSON-like value
			done(event.Result)

			// unsubscribe
			composer.Unsubscribe("hello-world", info.TransactionID, subscriberID)
		case "onStepFailure":
			slog.Error("Workflow failed", "transaction_id", event.TransactionID, "errors", event.Errors)
		}
	})

	return info.TransactionID, nil
}

In the above example, you execute the long-running workflow hello-world with the Notify option, then subscribe to its execution using its transaction ID.

Subscribe Function#

composer.Subscribe listens to changes in the workflow execution's status. It accepts the following parameters:

The EventType of the event is one of onBegin, onStepBegin, onStepSuccess, onStepFailure, onStepAwaiting, onCompensateBegin, onCompensateStepSuccess, onCompensateStepFailure, and onFinish.

If the EventType is onFinish, the workflow finished executing. The event then also has a Result property holding the workflow's output, and an Errors property with the errors that occurred, if any. The result is a JSON-like value: an object is a *dal.Record, not the Go type your workflow returns.

Unsubscribe Function#

You can unsubscribe from the workflow using composer.Unsubscribe, which accepts the workflow ID, the transaction ID, and the subscriber ID you passed to Subscribe.

Nested Long-Running Workflows#

When you execute a long-running workflow within another workflow, the parent workflow becomes a long-running workflow as well. The parent workflow waits for the nested workflow to finish before continuing its execution.

To change the status of an async step of the nested workflow, use the ID of the workflow you executed (the parent) as the WorkflowID of the IdempotencyKey, with the name of the nested workflow's step as the StepID.

Learn more about nested workflows in the Execute Nested Workflows chapter.

Workflow Engine Modules#

The Workflow Engine Module executes the workflows. MedusaGo comes with two:

• The in-memory Workflow Engine Module, which is the default. It keeps the steps that wait for their status in the memory of the process. If the process restarts, they're lost, and the executions stored in the database stay in their last state.
• The Redis Workflow Engine Module, which stores the checkpoints of executions, so a waiting step can be completed from any instance of your application. Use it in production. Learn how to configure it in the MedusaGo Configurations chapter.