Saga State Machines

The ability to coordinate a long-running business process across multiple messages and services is a core part of building distributed systems. MassTransit supports this with consumer sagas and saga state machines.

A saga is a long-lived transaction managed by a coordinator. It is started by an event, reacts to later events, stores progress between those events, and drives the next steps in the process by publishing events or sending commands. Sagas are designed to manage distributed workflows without locking resources or requiring immediate consistency.

We didn’t create it, we learned it from the original Princeton paper and from Arnon Rotem-Gal-Oz’s description.

Note

In MassTransit, sagas are a form of orchestration. For choreography, see Routing Slips.

Why use a saga?

Saga state machines are useful when a process:

• Spans multiple messages over time
• Depends on work done by other services
• Needs to remember prior decisions or data
• Must handle timeouts, retries, duplicate delivery, or out-of-order events
• May need to react differently depending on the current step in the process

Common examples include order processing, reservations, approvals, onboarding, and any workflow where the system is waiting for a later message before it can continue.

Saga state machines

MassTransit supports saga state machines, which provide a powerful syntax to create sagas. This is the preferred way to create sagas, as it is more expressive and easier to understand for long-running workflows.

A saga state machine models the workflow explicitly:

• The saga instance stores the current state and any process data
• States describe where the process currently is
• Events describe messages the saga can consume
• Behaviors define what happens when an event is received in a given state
• Transitions move the saga instance from one state to another

This makes the lifecycle of the workflow visible in code, which is why saga state machines are the preferred approach.

State machine behavior

At runtime, a saga state machine behaves like a stateful consumer:

1. A message arrives at the saga endpoint.
2. The message is correlated to an existing saga instance, or creates a new instance if the event is accepted in the Initial state.
3. The behavior for that event and the current state is executed.
4. The saga may update instance data, publish events, send commands, schedule timeouts, or finalize the workflow.
5. The updated saga instance is persisted so the next correlated message can continue from the correct point.

This is the key difference from a regular consumer: the saga remembers where the process is between messages.

State machine definition

Every saga state machine has two parts:

• A state machine class derived from MassTransitStateMachine<TInstance>
• An instance class implementing SagaStateMachineInstance

The instance stores the durable state for one in-flight workflow.

State machine use cases

Saga state machines are a good fit when a workflow needs to be explicit and recoverable. They help by:

• Recording process progress between messages
• Handling multiple valid event orders
• Coordinating commands, events, requests, and scheduled timeouts
• Recovering cleanly after service restarts or partial failures
• Keeping workflow rules in one place instead of scattering them across consumers

They are not a replacement for short, single-message handlers. If a process can be completed entirely inside one consumer without waiting for later messages, a consumer is usually simpler.

State machine building blocks

These concepts appear throughout saga state machine configuration:

• State: a named step in the workflow, such as Submitted or Accepted
• Event<T>: a message the saga can consume
• Initially: behavior for events handled while creating a new instance
• During(...): behavior for events handled in existing states
• TransitionTo(...): moves the instance to a new state
• Finalize(): marks the workflow as complete
• SetCompletedWhenFinalized(): removes finalized instances from the repository

Those pieces let you express the workflow as a series of state-specific reactions to messages.

Consumer sagas

MassTransit supports consumer sagas, which implement one or more interfaces to consume correlated saga events. This support is included so that it is straightforward to move applications from other saga implementations to MassTransit.

Consumer sagas can be enough for simple cases, but they do not present the workflow as clearly as a state machine.

Next steps

With an understanding of the basics, the next step is to learn how to:

1. Create a saga state machine
2. States, Events, and Behaviors
3. Configure the saga and choose a saga repository

Saga Persistence

Sagas are stateful event-based message consumers, so saving an instance between events is essential. Without persistence, each event would look like a brand new workflow and the later steps would have no context.

To store the saga state, you need to use one form of saga persistence. There are several types of storage that MassTransit supports. All of those, which are included in the main distribution, are listed below. There is also an in-memory unreliable storage, which allows temporarily storing your saga state. It is useful to try things out since it does not require any infrastructure.

Correlation

Correlation determines which saga instance should receive an event.

• The preferred approach is to correlate using the saga CorrelationId
• Query-based correlation is supported when the incoming event uses another business identifier
• Initial events that do not provide the final CorrelationId must define how a new identifier is chosen

Correct correlation is central to saga design. Every event on a saga state machine should have a clear and deterministic correlation strategy.

There are two types of saga repository:

• Query repository
• Identity-only repository

Depending on the persistence mechanism, repository implementation can be either identity-only or identity plus query.

When using identity-only repository, such as Azure Service Bus message session or Redis, you can only use correlation by identity. This means that all events that the saga receives must hold the saga correlation id, and the correlation for each event can only use CorrelateById method to define the correlation.

Query repository, by definition, supports identity correlation too, but in addition supports other properties of events being received and saga state properties. Such correlations are defined using CorrelateBy method, and you can use any logical expression that involve the event data and saga state data to establish such correlation. Repository implementations such as Entity Framework, NHibernate, and Marten support correlation by query. Of course, in-memory repository supports it as well.

Considerations

What follows is a set of guidelines related to sagas that was collected from Discord, Stack Overflow, and other sources to provide an easy way to link answers to commonly asked questions.

Identity

Saga instances are identified by a unique identifier (Guid), represented by the CorrelationId on the saga instance. Events are correlated to the saga instance using either the unique identifier or alternatively using an expression that correlates properties on the saga instance to each event. If the CorrelationId is used, it’s always a one-to-one match, either the saga already exists or it’s a new saga instance. With a correlation expression, the expression might match to more than one saga instance, so care should be used — because the event would be delivered to all matching instances.

Seriously, don’t send an event to all instances — unless you want to watch your message consumers lock your entire saga storage engine.

It is strongly advised to have CorrelationId as your table/document key. This will enable better concurrency handling and will make the saga state consistent.

Publishing and sending from sagas

Sagas are completely message-driven and therefore not only consume but also publish events and send commands. However, if your saga received a lot of messages coming roughly at the same time and the endpoint is set to process multiple messages in parallel — this can lead to a conflict between message processing and saga persistence.

This means that there could be more than one saga state updates that are being persisted at the same time. Depending on the saga repository type, this might fail for different reasons — versioning issue, row or table lock or eTag mismatch. All those problems are basically saying that you are having a concurrency issue.

It is normal for the saga repository to throw an exception in such a case, but if your saga is publishing messages, they were already published, but the saga state has not been updated. MassTransit will eventually use a retry policy on the endpoint, and more messages will be sent, potentially leading to mess. Or, if there are no retry policies configured, messages might be sent indicating that the process needs to continue. However, saga instance will be in the old state and will not accept any further messages because they will come in a wrong state.

This issue is common and can be solved by postponing the message publish and send operations until all persistence work is done. All messages that should be published are collected in a buffer, which is called an outbox.

Relational database recommendations

While it’s nice if you are developing a green-field system, and you can define your Saga Db Entity with CorrelationId as the Primary Key (Clustered), sometimes we have to work within existing db entities. If this is the case, please remember that to keep your sagas performing quickly (optimistic OR pessimistic, it doesn’t matter), you should follow the note below.

Tip

The CorrelationId should preferably be the Primary Key and Clustered for your saga table. If unable, then it must be a Clustered Index and Unique. And it’s also highly recommended to use the NewId package for creating nice Db Friendly guids.

Optimistic vs. pessimistic concurrency

Most persistence mechanisms for sagas supported by MassTransit need some way to guarantee ACID when processing sagas. Because there can be multiple threads consuming multiple bus events meant for the same saga instance, they could end up overwriting each other (race condition).

Relational databases can easily handle this by setting the transaction type to serializable or (page/row) locking. This would be considered as pessimistic concurrency.

Another way to handle concurrency is to have some attribute like version or timestamp, which updates every time a saga is persisted. By doing that, we can instruct the database only to update the record if this attribute matches between what we are trying to persist and what is stored in the database record we are trying to update.

This is a type of concurrency called optimistic concurrency. It doesn’t guarantee your unit of work with the database will succeed (must retry after these exceptions), but it also doesn’t block anybody else from working within the same database page (not locking the table/page).

So, which one should I use?

For almost every scenario, it is recommended using the optimistic concurrency, because most state machine logic should be fairly quick.

If the chosen persistence method supports optimistic concurrency, race conditions can be handled rather easily by specifying a retry policy for concurrency exceptions or using generic retry policy.

Concurrency optimizations

Saga concurrency issues happen, particularly when using optimistic concurrency. The most common reasons include:

• Simultaneous events correlating to the same instance, typically from multiple sources running in parallel
• Commands from the saga to consumers, where the consumer is quick and responds before the saga has finished processing the initiating event

There are certainly others, but anytime multiple events correlate to the same instance, concurrency issues are a concern. For that reason, the following baseline receive endpoint configuration is recommended as a starting point (tuning will depend upon the saga, repository, environment, etc.).

To configure the Receive endpoint directly:

services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<OrderStateMachine, OrderState>()
        .MongoDbRepository(r =>
        {
            r.Connection = "mongodb://127.0.0.1";
            r.DatabaseName = "orderdb";
        });

    x.UsingRabbitMq((context,cfg) =>
    {
        cfg.ReceiveEndpoint("saga-queue", e =>
        {
            const int ConcurrencyLimit = 20; // this can go up, depending upon the database capacity

            e.PrefetchCount = ConcurrencyLimit;

            e.UseMessageRetry(r => r.Interval(5, 1000));
            e.UseInMemoryOutbox();

            e.ConfigureSaga<OrderState>(context, s =>
            {
                var partition = s.CreatePartitioner(ConcurrencyLimit);

                s.Message<SubmitOrder>(x => x.UsePartitioner(partition, m => m.Message.OrderId));
                s.Message<OrderAccepted>(x => x.UsePartitioner(partition, m => m.Message.OrderId));
                s.Message<OrderCanceled>(x => x.UsePartitioner(partition, m => m.Message.OrderId));
            });
        });
    }
});

Alternatively, if you are using a saga definition:

public sealed class OrderStateSagaDefinition : SagaDefinition<OrderState>
{
    private const int ConcurrencyLimit = 20; // this can go up, depending upon the database capacity

    public OrderStateSagaDefinition()
    {
        // specify the message limit at the endpoint level, which influences
        // the endpoint prefetch count, if supported.
        Endpoint(e =>
        {
            e.Name = "saga-queue";
            e.PrefetchCount = ConcurrencyLimit;
        });
    }

    protected override void ConfigureSaga(IReceiveEndpointConfigurator endpointConfigurator, ISagaConfigurator<OrderState> sagaConfigurator)
    {
        endpointConfigurator.UseMessageRetry(r => r.Interval(5, 1000));
        endpointConfigurator.UseInMemoryOutbox();

        var partition = endpointConfigurator.CreatePartitioner(ConcurrencyLimit);

        sagaConfigurator.Message<SubmitOrder>(x => x.UsePartitioner(partition, m => m.Message.OrderId));
        sagaConfigurator.Message<OrderAccepted>(x => x.UsePartitioner(partition, m => m.Message.OrderId));
        sagaConfigurator.Message<OrderCanceled>(x => x.UsePartitioner(partition, m => m.Message.OrderId));
    }
}

This example uses message retry (Concurrency issues throw exceptions), the InMemoryOutbox (to avoid duplicate messages in the event of a concurrency failure). It uses a partitioner to limit the Receive endpoint to only one concurrent message for each OrderId (the partitioner uses hashing to meet the partition count).

The partitioner in this case is only for this specific Receive endpoint, multiple service instances (competing consumer) may still consume events for the same saga instance.