Compensation Function
In this chapter, you'll learn what a compensation function is and how to add it to a step.
What is a Compensation Function#
A compensation function rolls back or undoes changes made by a step when an error occurs in the workflow.
For example, if a step creates a record, the compensation function deletes the record when an error occurs later in the workflow.
By using compensation functions, you provide a mechanism that guarantees data consistency in your application and across systems.
How to add a Compensation Function?#
A compensation function is passed as a third parameter to the composer.CreateStep function.
For example, create the file src/workflows/hello-world.go with the following content:
// step1 has a compensation function, the third argument of CreateStep. It
// only runs if an error occurs throughout the workflow.
var step1 = composer.CreateStep("step-1",
func(ctx composer.StepExecutionContext, _ any) (composer.StepResponse[string], error) {
message := "Hello from step one!"
slog.Info(message)
return composer.NewStepResponse(message), nil
},
func(ctx composer.StepExecutionContext, _ any) error {
slog.Info("Oops! Rolling back my changes...")
return nil
})
Each step can have a compensation function. It receives the step's context and the data the step passed to it, and returns an error if it fails to undo the changes. The compensation function only runs if an error occurs throughout the workflow.
Test the Compensation Function#
Create a step in the same src/workflows/hello-world.go file that returns an error:
// step2 fails.
var step2 = composer.CreateStep("step-2",
func(ctx composer.StepExecutionContext, _ any) (composer.StepResponse[any], error) {
return composer.StepResponse[any]{}, common.NewMedusaError(
common.MedusaErrorTypes.UNEXPECTED_STATE,
"Throwing an error...",
)
}, nil)
Then, create a workflow that uses the steps:
// HelloWorldWorkflow runs step1, then step2, which fails. So, the
// compensation function of step1 runs.
var HelloWorldWorkflow = composer.CreateWorkflow("hello-world",
func(e *composer.Execution, _ any) (string, error) {
message, err := step1.Run(e, nil)
if err != nil {
return "", err
}
if _, err := step2.Run(e, nil); err != nil {
return "", err
}
return message, nil
})
Finally, execute the workflow from an API route, as explained in the Workflows chapter, and send a request to it.
In the application's logs, you'll see:
Hello from step one!, indicating that the first step ran successfully.Oops! Rolling back my changes..., indicating that the second step failed and the compensation function of the first step ran consequently.
The compensation functions run in the reverse order of the steps. The workflow's Run method returns the error of the step that failed.
nil as its data, since the step didn't return a response. Check the data before you use it.Pass Input to Compensation Function#
If a step creates a record, the compensation function must receive the ID of the record to remove it.
To pass input to the compensation function, pass a second parameter to composer.NewStepResponse in the step's response.
For example:
// stepWithInput passes data to its compensation function as the second
// argument of NewStepResponse.
var stepWithInput = composer.CreateStep("step-with-input",
func(ctx composer.StepExecutionContext, _ any) (composer.StepResponse[string], error) {
return composer.NewStepResponse(
"Hello from step one!",
map[string]string{"message": "Oops! Rolling back my changes..."},
), nil
},
func(ctx composer.StepExecutionContext, data any) error {
// data is nil if the step itself failed
input, ok := data.(map[string]string)
if !ok {
return nil
}
slog.Info(input["message"])
return nil
})
In this example, the step passes a map as a second parameter to NewStepResponse.
The compensation function receives it as its second parameter, with the type any. It uses a type assertion to get the map back, and uses its message key to log a message.
If you don't pass a second parameter, the compensation function receives the step's output instead.
*dal.Record values, arrays are []any, and numbers are float64. So, pass simple data to the compensation function, such as the ID of a record, and assert its type carefully.Resolve Resources from the Container#
The compensation function receives the step's context as a first parameter, and the data as a second one. The context has a Container property that you use to resolve resources from the container.
For example:
// stepWithContainer resolves resources from the container, which is in the
// context of the compensation function too.
var stepWithContainer = composer.CreateStep("step-with-container",
func(ctx composer.StepExecutionContext, _ any) (composer.StepResponse[string], error) {
return composer.NewStepResponse(
"Hello from step one!",
"Oops! Rolling back my changes...",
), nil
},
func(ctx composer.StepExecutionContext, data any) error {
logger := ctx.Container.Resolve(common.ContainerRegistrationKeys.LOGGER).(*slog.Logger)
message, _ := data.(string)
logger.Info(message)
return nil
})
In this example, you use the Container property of the context to resolve the logger. You then use the logger to log a message.
Handle Step Errors in Loops#
Consider you have a module that integrates a third-party ERP system, and you're creating a workflow that deletes items in that ERP. You may have a step that loops over the IDs of the items to retrieve each item's data, store it, then delete the item.
If an error occurs in the loop, the step returns the error without the response. So, its compensation function receives nil and can't restore the items that were already deleted.
To handle this, undo what the step did before you return the error. Keep what you delete in a variable, and restore it if a deletion fails:
// deleteItemsStep deletes items in the ERP. If deleting one of them fails,
// the step restores the items it deleted before returning the error, since
// the compensation function of a step that fails receives no data.
var deleteItemsStep = composer.CreateStep("delete-items-step",
func(ctx composer.StepExecutionContext, input DeleteItemsInput) (composer.StepResponse[[]string], error) {
erp := ctx.Container.Resolve("erp").(ERPService)
var previous []map[string]any
for _, id := range input.IDs {
data, err := erp.Retrieve(ctx.Context, id)
if err == nil {
err = erp.Delete(ctx.Context, id)
}
if err != nil {
// undo what the step did so far
for _, item := range previous {
_ = erp.Create(ctx.Context, item)
}
return composer.StepResponse[[]string]{}, err
}
previous = append(previous, data)
}
// the compensation function receives the deleted items
return composer.NewStepResponse(input.IDs, previous), nil
},
func(ctx composer.StepExecutionContext, data any) error {
erp := ctx.Container.Resolve("erp").(ERPService)
items, _ := data.([]map[string]any)
for _, item := range items {
if err := erp.Create(ctx.Context, item); err != nil {
return err
}
}
return nil
})
In this example, if deleting an item fails, the step restores the items it deleted so far, then returns the error. The error stops the workflow and triggers the compensation functions of the steps that ran before it.
If the step succeeds, it passes the items it deleted to the compensation function, which restores them if an error occurs later in the workflow.