Microservices with Dapr #7: Workflows

In this post I will focus on Dapr Workflows which can be a really useful solution for building multi-step processes in distributed systems.

Another extension of SavingOnDapr PoC project

This time we’ll be adding new functionality of “currency exchange” to introduce the concept behind Dapr Workflows with some practical example.

To recap: in SavingsOnDapr the main service of this system is API that exposes endpoints to manage two types of bank accounts (current and instant access savings). These accounts are organized in a form of hierarchy where savings account can be only created with a reference to an existing current account. We have the functionality of deposit transfers within the hierarchy implemented on Dapr Actor component. There is also EventStore service that is mostly used for the purpose of interest accrual for savings accounts (i.e. tracking balance changes within the time period).

Let’s say we now want to extend the system to offer currency exchange. There needs to be a way to specify account currency upon creation and the system is going to accept “currency exchange” orders. We don’t want to focus too much on how exchange rates are actually calculated and updated on regular basis but since there is some internal logic involved, it seems to make sense to have a separate service for that.

Our extended system could look something like this:

Dapr Workflow concept

Workflow building block has many analogs in other frameworks (Azure Durable Functions, NServiceBus or MassTransit Sagas - to name a few). They serve as a pattern for implementing long-running, multi-step business processes that span multiple services or modules in a distributed system. In such cases it is desired to have a high-level procedure (or state machine) that is responsible for business process orchestration.

Each step in such a process might represent individual activity or service call that varies in terms of complexity, retry policies, asynchronous or blocking nature etc. What’s important is that all these steps are connected by a shared state of the business process that should be persisted somewhere. When implementing it in a distributed system, we don’t want to make guesses at any point about the overall status by querying for details of individual steps.

You can read more about this concept and how it’s built into Dapr here: workflow overview. The important thing to be aware of in case of Dapr Workflows is that a lot of the complexity happens behind the scenes and that comes with some limitations on what could be done inside the workflow run.

Introducing CurrencyExchangeWorkflow

The main task here is adding a class that derives from Workflow<TInput, TOutput> abstract class. We need to provide an implementation of RunAsync(context, input) method which in our case looks as follows:

This method at first sight looks like a pretty standard piece of async/await C# code. One thing worth noting would be that all the “awaits” are called on a WorkflowContext instance. Another hint for what’s really going on would be reading the context methods that are called. We have following sequence:

1. CallActivityAsync() for ConfirmExchangeActivity

2. CreateTimer() and CallActivityAsync() in a loop until Status is different than Deferred…

3. CallActivityAsync() for DebitAccountActivity

4. WaitForExternalEventAsync() for AccountDebited event type (with timeout)

5. CallActivityAsync() for CreditAccountActivity

6. WaitForExternalEventAsync() for AccountCredited event type

So, the better way of thinking about Workflow.RunAsync() method is to treat it as an orchestrator for a complex process.

Each activity is a separate unit of work that might involve read/write operations against storage or service invocation via Dapr sidecar. Timer is a reminder that allows for off-loading the process at its current state and resuming it according to the schedule. Waiting for ExternalEvent works similarly but the resume trigger is an external event that can be raised for specific workflow instance.

It’s important to understand how workflow restores its internal execution state. It’s described quite well in Dapr Docs:

Workflow replay

Dapr Workflows maintain their execution state by using a technique known as event sourcing. Instead of storing the current state of a workflow as a snapshot, the workflow engine manages an append-only log of history events that describe the various steps that a workflow has taken. When using the workflow SDK, these history events are stored automatically whenever the workflow “awaits” for the result of a scheduled task.

When a workflow “awaits” a scheduled task, it unloads itself from memory until the task completes. Once the task completes, the workflow engine schedules the workflow function to run again. This second workflow function execution is known as a replay.

When a workflow function is replayed, it runs again from the beginning. However, when it encounters a task that already completed, instead of scheduling that task again, the workflow engine:

1. Returns the stored result of the completed task to the workflow.

2. Continues execution until the next “await” point.

Now it should be clear that the “await” inside a workflow method is not just an async call but rather some kind of a checkpoint that needs to behave in a deterministic manner in order to be “replayable”.

Here is how it looks like in the Postgres State Store (managed behind the scenes by Dapr). That’s the series of entries for a single “happy path” CurrencyExchange order:

The last entry contains base64 encoded ExchangeResult that is also exposed via API endpoint:

app.MapGet("v1/currency-exchange-order/{orderId}", 
           async (string orderId, DaprWorkflowClient wfClient) =>
{
   var state = await wfClient.GetWorkflowStateAsync(orderId);
   if (state is null)
   {
      return Results.NotFound();
   }
   else if (state.IsWorkflowCompleted)
   {
      return Results.Ok(
         new 
         {
            Status = Enum.GetName<WorkflowRuntimeStatus>(state.RuntimeStatus),
            Details = state.ReadOutputAs<ExchangeResult>()
         });
   }

   return Results.Accepted($"v1/currency-exchange-order/{orderId}");
}).WithTags(["currency-exchange"]);

http://localhost:5156/v1/currency-exchange-order/usd-eur-limitord-001
---------------------------------------------------------------------

{
  "status": "Completed",
  "details": {
    "succeeded": true,
    "message": "Exchange Order has been fulfilled",
    "receipt": {
      "targetAmount": 490.5,
      "exchangeRate": 0.95,
      "transactionDate": "2024-11-06T08:03:03.5226034Z",
      "transactionId": "usd-eur-limitord-001",
      "debtorExternalRef": "test-acc-usd-101",
      "beneficiaryExternalRef": "test-acc-10005",
      "sourceCurrency": "USD",
      "targetCurrency": "EUR"
    }
  }
}

Integrating Workflow into distributed system

Apart from the API endpoints for creating CurrencyExchange workflow and checking its result, integrating this class into AspNetCore web app required some additional steps. First, let’s look at the configuration in Program.cs:

builder.Services.AddDaprWorkflow(opts =>
{
    opts.RegisterWorkflow<CurrencyExchangeWorkflow>();

    opts.RegisterActivity<ConfirmExchangeActivity>();
    opts.RegisterActivity<DebitAccountActivity>();
    opts.RegisterActivity<CreditAccountActivity>();
});

Apart from the Workflow class, we need to register all of the WorkflowActivities (since these are not attached to the Workflow class, I would bet these can be reused for multiple workflows - which might be useful). Another thing we need to do is to establish some custom connection between domain events (Account Debited/Credited) and the workflow instances. This turned out to be a bit messy because of similar mechanism used for DepositTransferActor, but for now the PubSub event subscriptions are handled like that:

Everything each of these endpoints does is to extract the workflow instanceId (“messy” part) and call the DaprWorkflowClient.RaiseEventAsync() - which in turn causes this specific instance to be resumed from awaiting external event.

Dapr Workflows vs Actors

Now that we have both Workflows and Actors covered with examples of multi-step processes that have similar complexity, it is a good time to make some comparisons.

DepositTransferActor implementation can be found here. Although most of the logic was moved to separate service class, it’s still visible that Actor is more of a lightweight low-level building block. We have StateManager member that is needed for read/write interactions with Actor’s internal state and there is this IRemindable interface that enables Actor to receive reminder signals. There is quite a lot of flexibility regarding what is happening with Actor state but as a tradeoff there are many more things that we need to take care of on the implementation level.

For Dapr Workflows it’s quite the opposite situation. We have a lot of implementation details hidden out of our sight. This comes with some significant learning curve when we get to implement a single async method that is actually some sort of a state machine powered by internal event sourcing mechanism. Things are not what they seem initially… I guess it gets more interesting when trying to add some piece of logic that would be perfectly fine for a standard async method but is not compatible with the deterministic requirements of a workflow (some notes of such limitations can be found here). What might make this topic even more interesting is the fact that Dapr Workflow uses Actors internally (read more).

---

That’s all for now. Thanks for reading through! Code for this post can be found here.