Locking Operations in Workflows

In this chapter, you'll learn how to lock operations in your workflows to avoid race conditions and ensure data consistency.

What is a Lock?#

A lock is a mechanism that restricts multiple accesses to a resource or a piece of code, preventing multiple processes from modifying it simultaneously.

Locks are particularly useful in workflows where concurrent executions may lead to commerce risks. For example, MedusaGo uses locks in its cart workflows to prevent issues like overselling products or double charging customers.

The Locking Module handles the locking mechanism using the underlying provider. By default, it uses an in-memory provider, which only locks within the process of your application. For an application running multiple instances, use the Redis locking provider, as explained in the MedusaGo Configurations chapter.

You can use the Locking Module's service in your workflows to create locks around critical sections of your code.

How to Use Locks in Workflows#

MedusaGo provides two steps that you can use to create locks in your workflows. They are in the core-flows/src/locking/steps package of the MedusaGo repository:

• AcquireLockStep: attempt to acquire a lock. If the lock is already held by another process, this step waits until the lock is released. It fails if a timeout occurs before acquiring the lock.
• ReleaseLockStep: release a previously acquired lock.

You can use these steps in your workflows to ensure that only one instance of a workflow can modify a resource at a time.

For example:

// ChargeCustomerWorkflow locks the order while it charges the customer.
var ChargeCustomerWorkflow = composer.CreateWorkflow("charge-customer",
	func(e *composer.Execution, input WorkflowInput) (string, error) {
		_, err := lockingsteps.AcquireLockStep.Run(e, lockingsteps.AcquireLockStepInput{
			Key: []string{input.OrderID},
			// Attempt to acquire the lock for two seconds before timing out
			Timeout: 2,
			// Lock is only held for a maximum of ten seconds
			TTL: 10,
		})
		if err != nil {
			return "", err
		}

		result, err := chargeCustomerStep.Run(e, input)
		if err != nil {
			return "", err
		}

		_, err = lockingsteps.ReleaseLockStep.Run(e, lockingsteps.ReleaseLockStepInput{
			Key: []string{input.OrderID},
		})
		return result, err
	})

In this example, the workflow attempts to acquire a lock on an order using its ID as the lock key.

Once the lock is acquired, it executes the chargeCustomerStep. After the step is complete, it releases the lock using the ReleaseLockStep.

If the lock can't be acquired within the specified timeout period, the AcquireLockStep returns an error, and the workflow fails.

Locking Steps Input#

The input of AcquireLockStep is an AcquireLockStepInput value with the following properties:

• Key: the keys to lock, as a slice of strings.
• Timeout: the number of seconds to keep trying to acquire the lock. It's 0 by default, so the step fails if the lock is already held.
• RetryInterval: a pointer to the number of seconds to wait between attempts. It's 0.3 by default.
• TTL: the number of seconds after which the lock expires. It doesn't expire by default.
• OwnerID: a pointer to the ID of the lock's owner.
• Provider: the name of the locking provider to use. The default provider is used if it's empty.
• ExecuteOnSubWorkflow: whether to lock when the workflow runs as a step of another workflow. It's explained later in this chapter.

ReleaseLockStepInput accepts Key, OwnerID, Provider, and ExecuteOnSubWorkflow.

How to Use Locks in Workflow Steps#

You can alternatively acquire and release locks within steps by resolving the Locking Module's service and using its Acquire and Release methods.

For example:

// lockedChargeCustomerStep acquires and releases the lock itself, with the
// Locking Module's service.
var lockedChargeCustomerStep = composer.CreateStep("locked-charge-customer",
	func(ctx composer.StepExecutionContext, input WorkflowInput) (composer.StepResponse[string], error) {
		locking, ok := ctx.Container.Resolve(utilsmodulessdk.LOCKING).(lockingtypes.ILockingModule)
		if !ok {
			return composer.StepResponse[string]{}, errors.New("the Locking Module isn't registered")
		}

		err := locking.Acquire(ctx.Context, []string{input.OrderID}, lockingtypes.AcquireArgs{
			// Lock will auto-expire after 10 seconds
			Expire: 10,
		})
		if err != nil {
			return composer.StepResponse[string]{}, err
		}

		// TODO charge customer

		if _, err := locking.Release(ctx.Context, []string{input.OrderID}, lockingtypes.ReleaseArgs{}); err != nil {
			return composer.StepResponse[string]{}, err
		}
		return composer.NewStepResponse("charged"), nil
	}, nil)

In this example, the step acquires a lock on the order using the OrderID as the lock key. After performing the operation, it releases the lock.

The Execute method holds the lock while a function runs, and releases it when the function returns:

// chargeWithExecute holds the lock while a function runs. The lock is
// released when the function returns.
var chargeWithExecute = composer.CreateStep("charge-with-execute",
	func(ctx composer.StepExecutionContext, input WorkflowInput) (composer.StepResponse[string], error) {
		locking := ctx.Container.Resolve(utilsmodulessdk.LOCKING).(lockingtypes.ILockingModule)

		_, err := locking.Execute(ctx.Context, []string{input.OrderID}, func(ctx context.Context) (any, error) {
			// TODO charge customer
			return nil, nil
		}, lockingtypes.ExecuteArgs{Timeout: lockingtypes.Seconds(5)})
		if err != nil {
			return composer.StepResponse[string]{}, err
		}
		return composer.NewStepResponse("charged"), nil
	}, nil)

The Locking Module's service has the following methods:

• Acquire(ctx, keys, args): acquire the locks of the keys. args.Expire is the number of seconds after which a lock expires.
• Release(ctx, keys, args): release the locks. It returns whether they were released.
• Execute(ctx, keys, job, args): acquire the locks, run job, then release them. args.Timeout is the number of seconds to wait for the locks.
• ReleaseAll(ctx, args): release all the locks of an owner.

The step waits until the lock is acquired based on the configured Locking Module provider's options.

Locks in Nested Workflows#

When a nested workflow acquires a lock with AcquireLockStep, the lock isn't acquired. So, if you're using a workflow that acquires locks with AcquireLockStep within another workflow, the parent workflow must handle the locking mechanism. You can also set ExecuteOnSubWorkflow to true in the steps' input to lock even when the workflow runs as a step.

For example, consider a custom workflow that executes MedusaGo's CompleteCartWorkflow. The CompleteCartWorkflow acquires locks on the cart with AcquireLockStep to prevent race conditions.

Therefore, the custom workflow must acquire and release the locks around the CompleteCartWorkflow execution. For example:

// CustomCompleteCartWorkflow acquires the lock of the cart around the
// nested workflow, as the lock steps of a workflow run as a step don't
// lock.
var CustomCompleteCartWorkflow = composer.CreateWorkflow("custom-complete-cart",
	func(e *composer.Execution, input CustomCompleteCartInput) (*dal.Record, error) {
		// acquire the same lock as the nested workflow
		_, err := lockingsteps.AcquireLockStep.Run(e, lockingsteps.AcquireLockStepInput{
			Key:     []string{input.CartID},
			Timeout: 30,
			TTL:     60 * 2,
		})
		if err != nil {
			return nil, err
		}

		// execute the nested workflow
		result, err := workflows.CompleteCartWorkflow.RunAsStep(e, workflows.CompleteCartWorkflowInput{ID: input.CartID})
		if err != nil {
			return nil, err
		}

		// release the lock after the nested workflow is complete
		_, err = lockingsteps.ReleaseLockStep.Run(e, lockingsteps.ReleaseLockStepInput{
			Key: []string{input.CartID},
		})
		return result, err
	})

In the example above, the CustomCompleteCartWorkflow acquires a lock on the cart before executing the CompleteCartWorkflow and releases it afterward.

Make sure you're acquiring the same lock key as the nested workflow.