DEV Community: Jimmy Bogard

Life Beyond Distributed Transactions: An Apostate's Implementation - Relational Resources

Jimmy Bogard — Tue, 05 Feb 2019 20:46:04 +0000

Posts in this series:

So far in this series we've mainly concerned ourselves with a single resource that can't support distributed (or multi-entity) transactions. While that is becoming less common as NoSQL options, as Azure CosmosDB supports them, and with the 4.0 release, MongoDB now supports multi-document transactions.

What does this mean for us then?

Firstly, there are many cases that even though transactions between disparate entities is possible, it may not be desired. We may have to make design or performance compromises to make it work, as the technology to perform multi-entity transactions will always add some overhead. Even in SQL-land, we still often need to worry about consistency, locks, concurrency, etc.

The scenario that is most often overlooked I've found is bridging disparate resources, not multiple entities in the same resource (such as with many NoSQL databases). I've blogged about resilience patterns, especially crossing incorporating calling APIs to external sources. Most typically though, I see people trying to write to a database and send a message to a queue:

The code is rather innocuous - we're trying to do some stuff in a database, which IS transactional across multiple writes, but at the same time, try to write to a queue/broker/exchange like RabbitMQ/Azure Service Bus.

While it may be easy/obvious to spot places where we're trying to incorporate external services into our business transactions, like APIs, it's not so obvious when it's infrastructure we do own and is transactional.

Ultimately, our solution will be the same, which is to apply the outbox pattern.

Outbox pattern generalized

With the Cosmos DB approach, we placed the outbox inside each individual document. This is because our transactional boundary is a single item in a single collection. In order to provide atomicity for "publishing" messages, we need to design our outbox to the same transactional scope as our business data. If the transactional scope is a single record, the outbox goes in that record. If the scope is the database, the outbox goes in the database.

With a SQL database, our transaction scope widens considerably - to the entire database! This also widens the possibilities of how we can model our "outbox". We don't have to store an outbox per record in SQL Server - we can instead create a single outbox for the entire database.

What should be in this outbox table? Something similar to our original outbox in the Cosmos DB example:

Message ID (which message is this)
Type (what kind of message is this)
Body (what's in the message)

We also need to keep track of what's been dispatched or not. In the CosmosDB example, we keep sent/received messages inside each record.

With SQL though, having a single table that grows without bound is...problematic, so we need some more information about whether or not a message has been dispatched, as well as a dispatch date to be able to clean up after ourselves.

We can combine both into a single value however - "DispatchedAt", as a timestamp:

Processing the outbox will be a bit different than our original Cosmos DB example, since we now have a single table to deal with.

Sending via our SQL outbox

In the Cosmos DB example, we used domain events to communicate. We can do similar in our SQL example, but we'll more or less need to draw some boundaries. Are our domain events just to coordinate activities in our domain model, or are they there to interact with the outside world as service-level integration events? Previously, we only used them for domain events, and used special domain event handlers to "republish" as service events.

There are lots of tradeoffs between each approach, but in general, it's best not to combine the idea of domain events and service events.

To keep things flexible, we can simply model our messages as plainly as possible, to the point of just including them on the DbContext:

public Task SaveMessageAsync<T>(T message)  
{
    var outboxMessage = new OutboxMessage
    {
        Id = Guid.NewGuid(),
        Type = typeof(T).FullName,
        Body = JsonConvert.SerializeObject(message)
    };
    return Outbox.AddAsync(outboxMessage);
}

When we're doing something that needs any kind of messaging, instead of sending directly to a broker, we save a new message as part of a transaction:

using (var context = new AdventureWorks2016Context())  
{
    using (var transaction = context.Database.BeginTransaction())
    {
        var product = await context.Products.FindAsync(productId);

        product.Price = newPrice;
        await context.SaveMessageAsync(new ProductPriceChanged 
        {
            Id = productId, 
            Price = newPrice
        });
        await context.SaveChangesAsync();

        transaction.Commit();
    }
}

Each request that needs to send a message does so by only writing to the business data and outbox in a single transaction:

Then, as with our Cosmos DB example, a dispatcher reads from the outbox and sends these messages along to our broker:

And also like our Cosmos DB example, our dispatcher can run right after the transaction completes, as well as a background process to "clean up" after any missed messages. After each dispatch, we'd set the date of dispatch to ensure we skip already dispatched messages:

using (var context = new AdventureWorks2016Context())  
{
    using (var transaction = context.Database.BeginTransaction())
    {
        var message = await context.Outbox
             .Where(m => m.DispatchedAt == null)
             .FirstOrDefaultAsync();

        bus.Send(message);
        message.DispatchedAt = DateTime.Now;
        await context.SaveChangesAsync();

        transaction.Commit();
    }
}

With this in place, we can safely "send" messages in a transaction. However, we still have to deal with receiving these messages twice!

De-duplicating messages

In order to make sure we achieve at-least-once delivery, but exactly-once processing, we'll have to keep track of messages we've processed. To do so, we can just add a little extra information to our outbox - not just when we've sent messages, but when we've processed messages:

Similar to our inbox, we'll include the processed date with each message. As we process a message, we'll double-check our outbox to see if it's already processed before. If not, we can perform the work. If so, we just skip the message - nothing to be done!

With these measures in place, the last piece is to decide how long we should keep track of messages in our outbox. What's the longest amount of time a message can be marked as processed that we might receive the message again? An hour? A day? A week? Probably not a year, but something that makes sense. I'd start large, say a week, and move it lower as we understand the characteristics of our system.

With the outbox pattern, we can still coordinate activities between transactional resources by keeping track of what we need to communicate, when we've communicated, and when we've processed. It's like a little to-do list that our system uses to check things off as it goes, never losing track of what it needs to do.

In our last post, I'll wrap up and cover some scenarios where we should avoid such a level of coordination.

Life Beyond Distributed Transactions: An Apostate's Implementation - Sagas

Jimmy Bogard — Wed, 12 Sep 2018 15:25:04 +0000

Posts in this series:

Sample code from this series

So far in this series, we've looked at the ins and outs of moving beyond distributed transactions using persisted messages as a means of coordination between different documents (or resources). One common question in the example I give is "how do I actually make sure either both operations happen or neither?" To answer this question, we need to recognize that this "all-or-nothing" approach is a kind of transaction. But we've already said we're trying to avoid distributed transactions!

We won't be building a new kind of distributed transaction, but instead one that lives longer than any one single request, or a long-lived transaction. To implement a long-lived transaction, we need to look at the Saga pattern, first described in the original Sagas paper (Molina, Salem). The most common example of a saga I've seen described is booking a vacation. When you book a vacation, you need to:

Book a flight
Book a hotel
Reserve a car

You can't do all three at once - that's like getting a conference call together with all companies and getting consensus altogether. Not going to happen! Instead, we build this overall business transaction as a series of requests and compensating actions in case something goes wrong:

Cancel flight
Cancel hotel
Cancel car reservation

Our saga operations can be linear (controller pattern) or parallel (observer pattern) or in microservices terms, orchestration/choreography.

In order to satisfy our saga constraints, our requests must:

Be idempotent
Can abort

And the compensating requests must:

Be idempotent
Cannot abort
Be commutative

In our example, we have a model that basically assumes success. We:

Approve an order
Deduct stock

Let's modify this a bit to create an order fulfillment saga. For this saga, we can fulfill an order if and only if:

Our order is approved
We have enough stock

If our order is rejected, we need to release the stock. If we don't have enough stock, we need to un-approve (reject) our order. And keeping with our example, we need something to coordinate this activity - our saga. But rather than just call it some generic "saga", let's give it a meaningful name - OrderFulfillmentSaga:

This saga will coordinate the activities of the order request and stock. And because we need this saga to have the same communication properties of the other two documents, we can simply model this saga as just another document with our inbox/outbox!

The overall flow will be:

Once an order is created, kick off a new order fulfillment saga
This saga will coordinate actions between the stock and order request
If the order is rejected, the saga needs to initiate a stock return
If there is not enough stock, the saga needs to cancel the order

Let's start with kicking off the saga!

Kicking things off

When should we kick the saga off? It's tempting to do this in the initial request that creates a new order request, but remember - we can't put more than one document in a transaction unless we're very sure fulfillment and order requests will live close together in our Cosmos DB instance. That means we need to use our document messages to communicate with a saga - even if it doesn't exist!

We don't want to fulfill an order request twice, and for our simple scenario let's just assume an order request can't be "retried". Originally, our OrderRequest created an ItemPurchased document message for each item - we'll remove that in favor of a single OrderCreated document message:

public class OrderCreated : IDocumentMessage  
{
    public Guid Id { get; set; }
    public Guid OrderId { get; set; }

    public List<LineItem> LineItems { get; set; }

    public class LineItem
    {
        public int ProductId { get; set; }
        public int Quantity { get; set; }
    }
}

We could just have the OrderId and have the receiver then load up the OrderRequest, but for simplicity sake (and assuming you can't change the order after created), we'll treat this information as immutable and keep it in the message. Now when we create an OrderRequest, we'll also send this message:

public class OrderRequest : DocumentBase  
{
    public OrderRequest(ShoppingCart cart)
    {
        Id = Guid.NewGuid();
        Customer = new Customer
        {
            FirstName = "Jane",
            MiddleName = "Mary",
            LastName = "Doe"
        };
        Items = cart.Items.Select(li => new LineItem
        {
            ProductId = li.Key,
            Quantity = li.Value.Quantity,
            ListPrice = li.Value.ListPrice,
            ProductName = li.Value.ProductName
        }).ToList();
        Status = Status.New;

        Send(new OrderCreated
        {
            Id = Guid.NewGuid(),
            OrderId = Id,
            LineItems = Items
                .Select(item => new OrderCreated.LineItem
                {
                    ProductId = item.ProductId,
                    Quantity = item.Quantity
                })
                .ToList()
        });
    }

It's not much different than our original order creation - we're just now including the document message to initiate the order fulfillment saga.

Our handler for this document message needs to find the right OrderFulfillment saga document and let the saga handle the message:

public class OrderCreatedHandler : IDocumentMessageHandler<OrderCreated>  
{
    private readonly IDocumentDBRepository<OrderFulfillment> _repository;

    public OrderCreatedHandler(IDocumentDBRepository<OrderFulfillment> repository)
        => _repository = repository;

    public async Task Handle(OrderCreated message)
    {
        var orderFulfillment = (await _repository
                .GetItemsAsync(s => s.OrderId == message.OrderId))
            .FirstOrDefault();

        if (orderFulfillment == null)
        {
            orderFulfillment = new OrderFulfillment
            {
                Id = Guid.NewGuid(),
                OrderId = message.OrderId
            };

            await _repository.CreateItemAsync(orderFulfillment);
        }

        orderFulfillment.Handle(message);

        await _repository.UpdateItemAsync(orderFulfillment);
    }
}

Not shown here - but we do need to make sure we only have a single fulfillment saga per order, so we can configure inside Cosmos DB OrderId as a unique index.

The orderFulfillment.Handle method needs to start, and request stock:

public void Handle(OrderCreated message)  
{
    Process(message, m =>
    {
        if (IsCancelled)
            return;

        LineItems = m.LineItems
            .Select(li => new LineItem
            {
                ProductId = li.ProductId,
                AmountRequested = li.Quantity
            })
            .ToList();

        foreach (var lineItem in LineItems)
        {
            Send(new StockRequest
            {
                Id = Guid.NewGuid(),
                ProductId = lineItem.ProductId,
                AmountRequested = lineItem.AmountRequested,
                OrderFulfillmentId = Id
            });
        }
    });
}

In my example, I've made the OrderFulfillment saga coordinate with Stock with our StockRequest. This is instead of Stock listening for OrderCreated itself. My general thought here is that fulfillment manages the requests/returns for stock, and any business logic around that.

I also have a little check at the beginning - if an order is cancelled, we don't want to send out stock requests. This is the piece that's enforcing commutative requests - we might receive an order rejected notice before receiving the order created notice! When it comes to messaging, I always assume messages are received out of order, which means our business logic needs to be able to handle these situations.

Handling stock requests

Our original Stock implementation was quite naive, but this time we want to more intelligently handle orders. In our stock handler, we'll still have a document per product, but now it can make a decision based on the quantity available:

public void Handle(StockRequest message)  
{
    Process(message, e =>
    {
        if (QuantityAvailable >= message.AmountRequested)
        {
            QuantityAvailable -= e.AmountRequested;
            Send(new StockRequestConfirmed
            {
                Id = Guid.NewGuid(),
                OrderFulfillmentId = e.OrderFulfillmentId,
                ProductId = ProductId
            });
        }
        else
        {
            Send(new StockRequestDenied
            {
                Id = Guid.NewGuid(),
                OrderFulfillmentId = e.OrderFulfillmentId,
                ProductId = ProductId
            });
        }
    });
}

Because we're using document messages with our inbox de-duping messages, we don't need to worry about processing the stock request twice. Our simple logic just checks the stock, and if it's successful we can deduct the stock and return a StockRequestConfirmed message. If not, we can return a StockRequestDenied message.

A successful order fulfillment

Our original logic said that "an order can be fulfilled if the order is approved and we have enough stock". Approving an order is a human decision, so we have a basic form for doing so:

@if (Model.Order.Status == Status.New)
{
    <form asp-controller="Order" asp-action="Reject" asp-route-id="@Model.Order.Id" method="post">
        <input type="submit" value="Reject"/>
    </form>
    <form asp-controller="Order" asp-action="Approve" asp-route-id="@Model.Order.Id" method="post">
        <input type="submit" value="Approve"/>
    </form>
}

And when the order is approved, we just delegate to MediatR to handle this request:

public class ApproveOrder  
{
    public class Request : IRequest
    {
        public Guid Id { get; set; }
    }

    public class Handler : IRequestHandler<Request>
    {
        private readonly IDocumentDBRepository<OrderRequest> _orderRepository;

        public Handler(IDocumentDBRepository<OrderRequest> orderRepository)
        {
            _orderRepository = orderRepository;
        }

        public async Task<Unit> Handle(Request request, CancellationToken cancellationToken)
        {
            var orderRequest = await _orderRepository.GetItemAsync(request.Id);

            orderRequest.Approve();

            await _orderRepository.UpdateItemAsync(orderRequest);

            return Unit.Value;
        }
    }
}

Which then delegates to our document to approve the order request:

public void Approve()  
{
    if (Status == Status.Approved)
        return;

    if (Status == Status.Rejected)
        throw new InvalidOperationException("Cannot approve a rejected order.");

    Status = Status.Approved;
    Send(new OrderApproved
    {
        Id = Guid.NewGuid(),
        OrderId = Id
    });
}

We only want to send out the OrderApproved message once, so just some basic status checking handles that.

On the order fulfillment side:

public void Handle(OrderApproved message)  
{
    Process(message, m =>
    {
        OrderApproved = true;

        if (IsCancelled)
        {
            ProcessCancellation();
        }
        else
        {
            CheckForSuccess();
        }
    });
}

Each time we receive some external notification, we need to process the success/failure path, which I'll come back to in a bit. Our handler for StockRequestConfirmed will be similar, except we're tracking stock on a line item by line item basis:

public void Handle(StockRequestConfirmed message)  
{
    Process(message, m =>
    {
        var lineItem = LineItems.Single(li => li.ProductId == m.ProductId);
        lineItem.StockConfirmed = true;

        if (IsCancelled)
        {
            ReturnStock(lineItem);
        }
        else
        {
            CheckForSuccess();
        }
    });
}

The CheckForSuccess method will look to see if all the order fulfillment requirements are met:

private void CheckForSuccess()  
{
    if (IsCancelled)
        return;

    if (LineItems.All(li => li.StockConfirmed) && OrderApproved)
    {
        Send(new OrderFulfillmentSuccessful
        {
            Id = Guid.NewGuid(),
            OrderId = OrderId
        });
    }
}

Only if all of our stock has been confirmed and our order has been approved will we send a message back to the Order document to then finally complete the order:

public void Handle(OrderFulfillmentSuccessful message)  
{
    Process(message, m =>
    {
        if (Status == Status.Rejected || Status == Status.Cancelled)
            return;

        Status = Status.Completed;
    });
}

The overall message flow looks something like this:

For each step along the way, we've got idempotency handled for us by the inbox/outbox structures. However, we still need to handle out-of-order messages, which is why you'll see success/fail checks every time we receive a notification.

Now that we've got the success path taken care of, let's look at the failure paths.

Cancelling the order fulfillment

The first way our order fulfillment can be cancelled is if an order is rejected. From the web app, our Order document handles a rejection:

public void Reject()  
{
    if (Status == Status.Rejected)
        return;

    if (Status == Status.Approved)
        throw new InvalidOperationException("Cannot reject an approved order.");

    if (Status == Status.Approved)
        throw new InvalidOperationException("Cannot reject a completed order.");

    Status = Status.Rejected;
    Send(new OrderRejected
    {
        Id = Guid.NewGuid(),
        OrderId = Id
    });
}

Our order sends an OrderRejected document message that our order fulfillment document receives:

public void Handle(OrderRejected message)  
{
    Process(message, m =>
    {
        OrderRejected = true;

        Cancel();
    });
}

The Cancel method marks the order fulfillment as cancelled and then processes the cancellation:

private void Cancel()  
{
    IsCancelled = true;

    ProcessCancellation();
}

Similarly, a notification of StockRequestDenied will cancel the order fulfillment:

public void Handle(StockRequestDenied message)  
{
    Process(message, m =>
    {
        Cancel();
    });
}

In order to process our order fulfillment cancellation, we need to do a couple of things. First, we need to notify our Order document that it needs to be cancelled. And for any Stock items that were fulfilled, we need to return that stock:

private void ProcessCancellation()  
{
    if (!CancelOrderRequested && !OrderRejected)
    {
        CancelOrderRequested = true;
        Send(new CancelOrderRequest
        {
            Id = Guid.NewGuid(),
            OrderId = OrderId
        });
    }

    foreach (var lineItem in LineItems.Where(li => li.StockConfirmed))
    {
        ReturnStock(lineItem);
    }
}

Each step along the way, we keep track of what messages we've sent out so that we don't send notifications twice. To return stock:

private void ReturnStock(LineItem lineItem)  
{
    if (lineItem.StockReturnRequested)
        return;

    lineItem.StockReturnRequested = true;
    Send(new StockReturnRequested
    {
        Id = Guid.NewGuid(),
        ProductId = lineItem.ProductId,
        AmountToReturn = lineItem.AmountRequested
    });
}

If stock item has already had a return requested, we just skip it. Finally, the order can receive the cancel order request:

public void Handle(CancelOrderRequest message)  
{
    Process(message, m =>
    {
        if (Status == Status.Rejected)
            return;

        Status = Status.Cancelled;
    });
}

With our failure flow in place, the message flows looks something like:

Our order fulfillment saga can now handle the complex process of managing stock and order approvals, keeping track of each step along the way and dealing with success/failure when it receives the notifications. It handles idempotency, retries, and commutative/out-of-order messages.

In the next post, we'll look at how we can implement the inbox/outbox pattern for other resources, allowing us to bridge to other kinds of databases where a distributed transaction is just plain impossible.

Life Beyond Distributed Transactions: An Apostate's Implementation - Dispatcher Failure Recovery

Jimmy Bogard — Thu, 30 Aug 2018 13:15:14 +0000

Posts in this series:

Sample code from this series

In the last post, we looked at how we can recover from exceptions from inside our code handling messages. We perform some action in our document, and something goes wrong. But what happens when something goes wrong during the dispatch process:

If our dispatcher itself fails, either:

Pulling a message from the outbox
Sending a message to the receiver
Saving documents

Then our documents are still consistent, but we've lost the execution flow to dispatch. We've mitigated our failures somewhat, but we still can have the possibility of some unrecoverable failure in our dispatcher, and no amount of exception handling can prevent a document with a message sitting in its outbox, waiting for processing.

If we were able to wrap everything, documents and queues, in a transaction, then we could gracefully recover. However, the point of this series is that we don't have access to distributed transactions, so that option is out.

What we need is some kind of background process looking for documents with pending messages in their outbox, ready to process.

Designing the dispatch rescuer

We already have the dispatch recovery process using async messaging to retry an individual dispatch, which works great when the failure is in application code. When the failure is environmental, our only clue something is wrong is a document with messages in their outbox.

The general process to recover from these failures would be:

Find any documents with unprocessed messages (ideally oldest first)
Retry them one at a time

We have the possibility though that we have:

In flight dispatching
In flight retries

Ideally, we have some sort of lagging processor, so that when we have issues, we don't interfere with normal processing. Luckily for us, Cosmos DB already comes with the ability to be notified that documents have been changed, the Change Feed, and this change feed even lets us work with a built-in delay. After each document changes, we can wait some amount of time where we assume that dispatching happened, and re-check the document to make sure dispatching occurred.

Our rescuer will:

Get notified when a document changes
Check to see if there are still outbox messages to process
Send a message to reprocess that document

It's somewhat naive, as we'll get notified for all document changes. To make our lives a little bit easier, we can turn off immediate processing for document message dispatching and just dispatch through asynchronous processes, but it's not necessary.

Creating the change feed processor

Using the documentation as our guide, we need to create two components:

A document feed observer to receive document change notifications
A change feed processor to host and invoke our observer

Since we already have a background processor in our dispatcher, we can simply host the observer in the same endpoint. The observer won't actually be doing the work, however - we'll still send a message out to process the document. This is because NServiceBus still provides all the logic around retries and poison messages that I don't want to code again. Like most of my integrations, I kick out the work into a durable message and NServiceBus as quickly as possible.

That makes my observer pretty small:

public class DocumentFeedObserver<T> : IChangeFeedObserver  
    where T : DocumentBase
{
    static ILog log = LogManager.GetLogger<DocumentFeedObserver<T>>();

    public Task OpenAsync(IChangeFeedObserverContext context) 
        => Task.CompletedTask;

    public Task CloseAsync(
        IChangeFeedObserverContext context, 
        ChangeFeedObserverCloseReason reason)
        => Task.CompletedTask;

    public async Task ProcessChangesAsync(
        IChangeFeedObserverContext context, 
        IReadOnlyList<Document> docs, 
        CancellationToken cancellationToken)
    {
        foreach (var doc in docs)
        {
            log.Info($"Processing changes for document {doc.Id}");

            var item = (dynamic)doc;

            if (item.Outbox.Count > 0)
            {
                ProcessDocumentMessages message = ProcessDocumentMessages.New<T>(item);

                await Program.Endpoint.SendLocal(message);
            }
        }
    }
}

The OpenAsync and CloseAsync methods won't do anything, all my logic is in the ProcessChangesAsync method. In that method, I get a collection of changed documents. I made my DocumentChangeObserver generic because each observer observes only one collection, so I have to create distinct observer instances per concrete DocumentBase type.

In the method, I loop over all the documents passed in and look to see if the document has any messages in the outbox. If so, I'll create a new ProcessDocumentMessages to send to myself (as I'm also hosting NServiceBus in this application), which will then process the document messages.

With our simple observer in place, we need to incorporate the observer in our application startup.

Configuring the observer

For our observer, we have a couple of choices on how we want to process document changes. Because our observer will get called for every document change, we want to be careful about the work it does.

Our original design had document messages dispatched in the same request as the original work. If we keep this, we want to make sure that we minimize the amount of rework for a document with messages. Ideally, our observer only kicks out messages when there is truly something wrong with dispatching. This will also minimize the amount of queue messages, reserving them for the error case.

So a simple solution would be to just introduce some delay in our processing:

private static ChangeFeedProcessorBuilder CreateBuilder<T>(DocumentClient client)  
    where T : DocumentBase
{
    var builder = new ChangeFeedProcessorBuilder();
    var uri = new Uri(CosmosUrl);
    var dbClient = new ChangeFeedDocumentClient(client);

    builder
        .WithHostName(HostName)
        .WithFeedCollection(new DocumentCollectionInfo
        {
            DatabaseName = typeof(T).Name,
            CollectionName = "Items",
            Uri = uri,
            MasterKey = CosmosKey
        })
        .WithLeaseCollection(new DocumentCollectionInfo
        {
            DatabaseName = typeof(T).Name,
            CollectionName = "Leases",
            Uri = uri,
            MasterKey = CosmosKey
        })
        .WithProcessorOptions(new ChangeFeedProcessorOptions
        {
            FeedPollDelay = TimeSpan.FromSeconds(15),
        })
        .WithFeedDocumentClient(dbClient)
        .WithLeaseDocumentClient(dbClient)
        .WithObserver<DocumentFeedObserver<T>>();

    return builder;
}

The ChangeFeedProcessorBuilder is configured for every document type we want to observe, with a timespan in this example of 15 seconds. I could bump this up a bit - say to an hour or so. It will really depend on the business, the SLAs they expect for work to complete.

Finally, in our application startup, we need to create the builder, processor, and start it all up:

Endpoint = await NServiceBus.Endpoint.Start(endpointConfiguration)  
    .ConfigureAwait(false);

var builder = CreateBuilder<OrderRequest>(client);  
var processor = await builder.BuildAsync();

await processor.StartAsync();

Console.WriteLine("Press any key to exit");  
Console.ReadKey();

await Endpoint.Stop()  
    .ConfigureAwait(false);

await processor.StopAsync();

With this in place, we can have a final guard against failures, assuming that someone completely pulled the plug on our application and all we have left is a document with messages sitting in its outbox.

In our next post, we'll look at using sagas to coordinate changes between documents - what happens if we want either all, or none of our changes to be processed in our documents?

Life Beyond Distributed Transactions: An Apostate's Implementation - Failures and Retries

Jimmy Bogard — Thu, 16 Aug 2018 15:05:49 +0000

Posts in this series:

Sample code from this series

In the last post, we looked at an example of dispatching document messages to other documents using a central dispatcher. Our example worked well in the happy path scenario, but what happens when something goes wrong? We of course do not want a failure in dispatching messages to make the entire request fail, but what would that mean for us?

We described a general solution to put the dispatching work aside using queues and messaging, effectively saying "yes, dispatching failed, so let's put it aside to look at in the future". This would allow the overall main request to complete:

Our original example also assumed that we would dispatch our messages immediately in the context of the same request, which isn't a bad default but maybe isn't always desirable. Let's first look at the scenario of dispatching immediately, and what failures could mean.

Characterizing our failures

Dispatching failures could happen for a number of reasons, but I generally see a continuum:

Transient
Delayed
Permanent

My failures usually have some sort of time component associated with them. A transient failure may mean that if I simply try the action again immediately, it may work. This most often comes up with some sort of concurrency violation against the database.

Delayed failures are a bit different, where I won't succeed if I try immediately, but I will if I just wait some amount of time.

Permanent failures mean there's an unrecoverable failure, and no amount of retries will allow the operation to succeed.

Of course, we could simply ignore failures, but our business and customers might not agree with that approach. How might we handle each of these kinds of failures?

Transient failures

If something goes wrong, can we simply retry the operation? That seems fairly straightforward - but we don't want to retry too many times. We can implement some simple policies, either with a hardcoded number of retries or using something like the Polly Project to retry an action.

To keep things simple, we can have a policy to address the most common transient failure - optimistic concurrency problems. We first want to enable OCC checks, of course:

public async Task<Document> UpdateItemAsync(T item)  
{
    var ac = new AccessCondition
    {
        Condition = item.ETag,
        Type = AccessConditionType.IfMatch
    };

    return await _client.ReplaceDocumentAsync(
        UriFactory.CreateDocumentUri(DatabaseId, CollectionId, item.Id.ToString()),
        item,
        new RequestOptions { AccessCondition = ac });
}

When we get a concurrency violation, this results in a DocumentClientException with a special status code (hooray HTTP!). We need some way to wrap our request and retry if necessary - time for another MediatR behavior! This one will retry our action some number of times:

public class RetryUnitOfWorkBehavior<TRequest, TResponse>  
    : IPipelineBehavior<TRequest, TResponse>
{
    private readonly IUnitOfWork _unitOfWork;

    public RetryUnitOfWorkBehavior(IUnitOfWork unitOfWork) 
        => _unitOfWork = unitOfWork;

    public Task<TResponse> Handle(
        TRequest request, 
        CancellationToken cancellationToken,
        RequestHandlerDelegate<TResponse> next)
    {
        var retryCount = 0;

        while (true)
        {
            try
            {
                return next();
            }
            catch (DocumentClientException e)
            {
                if (e.StatusCode != HttpStatusCode.PreconditionFailed)
                    throw;

                if (retryCount >= 5)
                    throw;

                _unitOfWork.Reset();

                retryCount++;
            }
        }
    }
}

If our action files due to a concurrency problem, we need to clear out our unit of work's identity map and try again:

public void Reset()  
{
    _identityMap.Clear();
}

Then we just need to register our behavior with the container like our original unit of work behavior, and we're set. We could have of course modified our original behavior to add retries - but I want to keep them separate because they truly are different concerns.

That works for immediate failures, but we still haven't looked at failures in our message dispatching. For that, we'll need to involve some messaging.

Deferred dispatching with messaging and NServiceBus

The immediate retries can take care of transient failures during a request, but if the there's some deeper issue, we want to defer the document message dispatching to some time in the future. To make my life easier, and not have to implement half the Enterprise Integration Patterns book myself, I'll leverage NServiceBus to manage my messaging.

Our original dispatcher looped through our unit of work's identity map to find documents with messages that need dispatching. We'll want to augment that behavior to catch any failures, and dispatch those messages offline:

public interface IOfflineDispatcher  
{
    Task DispatchOffline(DocumentBase document);
}

Our Complete method of the unit of work will now take these failed dispatches and instruct our offline dispatcher to dispatch these offline:

public async Task Complete()  
{
    var toSkip = new HashSet<DocumentBase>(DocumentBaseEqualityComparer.Instance);

    while (_identityMap
            .Except(toSkip, DocumentBaseEqualityComparer.Instance)
        .Any(a => a.Outbox.Any()))
    {
        var document = _identityMap
            .Except(toSkip, DocumentBaseEqualityComparer.Instance)
            .FirstOrDefault(a => a.Outbox.Any());

        if (document == null)
            continue;

        var ex = await _dispatcher.Dispatch(document);
        if (ex != null)
        {
            toSkip.Add(document);

            await _offlineDispatcher.DispatchOffline(document);
        }
    }
}

This is a somewhat naive implementation - it doesn't allow for partial document message processing. If a document has 3 messages, we mark the retry the entire document instead of an individual message at at time. We could manage this more granularly, by including a "retry" collection on our document. But this introduces more issues - we could still have some system failure after dispatch and our document message never make it to retry.

When our transaction scope is individual operations instead of the entire request, we have to assume failure at every instance and examine what might go wrong.

The offline dispatcher uses NServiceBus to send a durable message out:

public class UniformSessionOfflineDispatcher  
    : IOfflineDispatcher
{
    private readonly IUniformSession _uniformSession;

    public UniformSessionOfflineDispatcher(IUniformSession uniformSession)
        => _uniformSession = uniformSession;

    public Task DispatchOffline(DocumentBase document)
        => _uniformSession.Send(ProcessDocumentMessages.New(document));
}

The IUniformSession piece from NServiceBus to send a message from any context (in a web application, backend service, etc.). Our message just includes the document ID and type:

public class ProcessDocumentMessages : ICommand  
{
    public Guid DocumentId { get; set; }
    public string DocumentType { get; set; }

    // For NSB
    public ProcessDocumentMessages() { }

    private ProcessDocumentMessages(Guid documentId, string documentType)
    {
        DocumentId = documentId;
        DocumentType = documentType;
    }

    public static ProcessDocumentMessages New<TDocument>(
        TDocument document)
        where TDocument : DocumentBase
    {
        return new ProcessDocumentMessages(
            document.Id, 
            document.GetType().AssemblyQualifiedName);
    }
}

We can use this information to load our document from the repository. With this message in place, we now need the component that will receive our message. For this, it will really depend on our deployment, but for now I'll just make a .NET Core console application that includes our NServiceBus hosting piece and a handler for that message:

I won't dig too much into the NServiceBus configuration as it's really not that germane, but let's look at the handler for that message:

public class ProcessDocumentMessagesHandler  
    : IHandleMessages<ProcessDocumentMessages>
{
    private readonly IDocumentMessageDispatcher _dispatcher;

    public ProcessDocumentMessagesHandler(IDocumentMessageDispatcher dispatcher) 
        => _dispatcher = dispatcher;

    public Task Handle(ProcessDocumentMessages message, IMessageHandlerContext context)
        => _dispatcher.Dispatch(message);
}

Also not very exciting! This class is what NServiceBus dispatches the durable message to. For our simple example, I'm using RabbitMQ, so if something goes wrong our message goes into a queue:

Our handler receives this message to process it. The dispatcher is slightly different, as it needs to work with a message instead of an actual document, so it needs to load it up first:

public async Task Dispatch(ProcessDocumentMessages command)  
{
    var documentType = Type.GetType(command.DocumentType);
    var repository = GetRepository(documentType);
    var document = await repository.FindById(command.DocumentId);

    if (document == null)
    {
        return;
    }

    foreach (var message in document.Outbox.ToArray())
    {
        var handler = GetHandler(message);

        await handler.Handle(message, _serviceFactory);

        document.ProcessDocumentMessage(message);

        await repository.Update(document);
    }
}

One other key difference in this dispatch method is that we don't wrap anything in any kind of try-catch to report back errors. In an in-process dispatch mode, we still want the main request to succeed. In the offline processing mode, we're only dealing with document dispatching. And since we're using NServiceBus, we can rely on its built-in recoverability behavior with immediate and delayed retries, eventually moving messages to an error queue.

With this in place, we can put forth an optimistic, try-immediately policy, but fall back on durable messaging if something goes wrong with our immediate dispatch. It's not bulletproof, and in the next post, I'll look at how we can implement some sort of doomsday scenario, where we have a failure between dispatch failure and queuing the retry message.

Life Beyond Distributed Transactions: An Apostate's Implementation - Dispatching Example

Jimmy Bogard — Mon, 13 Aug 2018 19:24:06 +0000

Posts in this series:

Sample code from this series

In the last post, we looked at refactoring our documents to use messaging to communicate changes. We're still missing something, however - the dispatcher:

Our dispatcher is the main component that facilitates document communication. For a given document, it needs to:

Read messages out of a document's outbox
Find the the document message handler for each, and invoke
Manage failures for a document message handler

We'll tackle that last piece in a future post. There's one piece we do need to think about first - where does the dispatcher get its list of documents to dispatch messages to?

Before we get to the dispatcher, we need to solve for this problem - knowing which documents need dispatching!

Introducing a unit of work

For a given request, we'll load up a document and affect some change with it. We already have a pinch point in which our documents are loaded - the repository. If we want to dispatch document messages in the same request, we'll need to keep track of our documents that we've loaded in a request. For this, we can use a Unit of Work.

Any ORM that you use will implement this pattern - for Entity Framework, for example, the DbContext is your Unit of Work. For Cosmos DB's SDK, there really isn't a concept of these ORM patterns. We have to introduce them ourselves.

Our unit of work will keep track of documents for a given session/request, letting us interact with the loaded documents during the dispatch phase of a request. Our Unit of Work will also serve as an identity map - the thing that makes sure that when we load a document in a request, it's only loaded once. Here's our basic IUnitOfWork interface:

public interface IUnitOfWork  
{
    T Find<T>(Guid id) where T : DocumentBase;
    void Register(DocumentBase document);
    void Register(IEnumerable<DocumentBase> documents);
    Task Complete();
}

The implementation contains the "identity map" as a simple HashSet

public class UnitOfWork : IUnitOfWork  
{
    private readonly ISet<DocumentBase> _identityMap 
        = new HashSet<DocumentBase>(DocumentBaseEqualityComparer.Instance);

Then we can register an instance with our UnitOfWork:

public void Register(DocumentBase document)  
{
    _identityMap.Add(document);
}

public void Register(IEnumerable<DocumentBase> documents)  
{
    foreach (var document in documents)
    {
        Register(document);
    }
}

Finding an existing DocumentBase just searches our identity map:

public T Find<T>(Guid id)  
    where T : DocumentBase
    => _identityMap.OfType<T>().FirstOrDefault(ab => ab.Id == id);

We'll come back to the Complete method, because this will be the part where we dispatch. We still need the part where we register our documents in the unit of work, and this will be in our repository implementation:

public async Task<T> GetItemAsync(Guid id)  
{
    try
    {
        var root = _unitOfWork.Find<T>(id);

        if (root != null)
            return root;

        Document document = await _client.ReadDocumentAsync(UriFactory.CreateDocumentUri(DatabaseId, CollectionId, id.ToString()));
        var item = (T)(dynamic)document;

        _unitOfWork.Register(item);

        return item;
    }
    catch (DocumentClientException e)
    {
        if (e.StatusCode == System.Net.HttpStatusCode.NotFound)
        {
            return null;
        }

        throw;
    }
}

We'll repeat this for any method in our repository that loads a document, registering and looking up in our unit of work.

With a means to track our documents, let's see how we'll dispatch.

Dispatching document messages

Our dispatcher's fairly straightforward - the only wrinkle is we'll need to surface any potential exception out. Instead of just crashing in case something goes awry, we'll want to just surface the exception and let the caller decide how to handle failures:

public interface IDocumentMessageDispatcher  
{
    Task<Exception> Dispatch(DocumentBase document);
}

If I'm dispatching a document message to three handlers, I don't want one handler prevent dispatching to others.

We have another challenge - our interface is not generic for dispatching, but the handlers and repositories are! We'll have to do some generics tricks to unwrap our base type to the correct generic types. The basic flow will be:

For each document message:
Find document message handlers
Call the handler
Remove the document message from the outbox
Save the document

Here's our basic implementation:

public async Task<Exception> Dispatch(DocumentBase document)  
{
    var repository = GetRepository(document.GetType());
    foreach (var documentMessage in document.Outbox.ToArray())
    {
        try
        {
            var handler = GetHandler(documentMessage);

            await handler.Handle(documentMessage, _serviceFactory);

            document.ProcessDocumentMessage(documentMessage);

            await repository.Update(document);
        }
        catch (Exception ex)
        {
            return ex;
        }
    }
    return null;
}

We first build a repository based on the document type. Next, we loop through each document message in the outbox. For each document message, we'll find the handler(s) and call them. Once those succeed, we'll process our document message (removing it from the outbox) and update our document. We want to update for each document message in the outbox - if there's 3 document messages in the outbox, we save 3 times to make sure once message completes we don't have to go back to it if something goes wrong.

The GetHandler method is a bit wonky, because we're bridging generics. Basically, we create a non-generic version of the document message handlers:

private abstract class DomainEventDispatcherHandler  
{
    public abstract Task Handle(
        IDocumentMessage documentMessage, 
        ServiceFactory factory);
}

Then create a generic version that inherits from this:

private class DomainEventDispatcherHandler<T> : DomainEventDispatcherHandler  
    where T : IDocumentMessage
{
    public override Task Handle(IDocumentMessage documentMessage, ServiceFactory factory)
    {
        return HandleCore((T)documentMessage, factory);
    }

    private static async Task HandleCore(T domainEvent, ServiceFactory factory)
    {
        var handlers = factory.GetInstances<IDocumentMessageHandler<T>>();
        foreach (var handler in handlers)
        {
            await handler.Handle(domainEvent);
        }
    }
}

I've used this pattern countless times, basically to satisfy the compiler. I've tried dynamic too but it introduces other problems. Then to call this, our GetHandler instantiates the generic version, but returns the non-generic base class:

private static DomainEventDispatcherHandler GetHandler(  
    IDocumentMessage documentMessage)
{
    var genericDispatcherType = typeof(DomainEventDispatcherHandler<>)
        .MakeGenericType(documentMessage.GetType());

    return (DomainEventDispatcherHandler)
        Activator.CreateInstance(genericDispatcherType);
}

With this, I can have non-generic code still call into generics. I'll do something similar with the repository:

private abstract class DocumentDbRepo  
{
    public abstract Task<DocumentBase> FindById(Guid id);
    public abstract Task Update(DocumentBase document);
}

With these bridges in place, my dispatcher can interact with the concrete generic repositories and handlers. The final piece is the document cleaning up its outbox:

public void ProcessDocumentMessage(  
    IDocumentMessage documentMessage)
{
    _outbox?.Remove(documentMessage);
}

With our dispatcher done, and our unit of work in place, we can now focus on the piece that will invoke our unit of work.

Building a MediatR behavior

We want our unit of work to complete with each request once everything is "done". For ASP.NET Core applications, this might mean some kind of filter. For us, I want the dispatching to work really with any context, so one possibility is to use a MediatR behavior to wrap our MediatR handler. A filter would work too of course, but we'd need to mimic our filters in tests if we want everything to still get dispatched appropriately.

The behavior is pretty straightforward:

public class UnitOfWorkBehavior<TRequest, TResponse>  
    : IPipelineBehavior<TRequest, TResponse>
{
    private readonly IUnitOfWork _unitOfWork;

    public UnitOfWorkBehavior(IUnitOfWork unitOfWork)
    {
        _unitOfWork = unitOfWork;
    }

    public async Task<TResponse> Handle(
        TRequest request, 
        CancellationToken token, 
        RequestHandlerDelegate<TResponse> next)
    {
        var response = await next();

        await _unitOfWork.Complete();

        return response;
    }
}

We do the main work, then once that's finished, complete our unit of work.

That's all of our infrastructure pieces, and the last part is registering these components with the DI container at startup:

services.AddMediatR(typeof(Startup));

services.AddScoped(typeof(IDocumentDBRepository<>), typeof(DocumentDBRepository<>));  
services.AddScoped<IUnitOfWork, UnitOfWork>();  
services.AddScoped<IDocumentMessageDispatcher, DocumentMessageDispatcher>();  
services.AddScoped(typeof(IPipelineBehavior<,>), typeof(UnitOfWorkBehavior<,>));  
services.Scan(c =>  
{
    c.FromAssembliesOf(typeof(Startup))
        .AddClasses(t => t.AssignableTo(typeof(IDocumentMessageHandler<>)))
        .AsImplementedInterfaces()
        .WithTransientLifetime();
});

We add our MediatR handlers using the MediatR.Extensions.Microsoft.DependencyInjection package, our generic repository, unit of work, dispatcher, and unit of work behavior. Finally, we add all of the IDocumentMessageHandler implementations using Scrutor, making our lives much easier to add all the handlers in one go.

With all this in place, we can run and verify that our handlers fire and we can see the message in the inbox of the Stock item:

{
    "QuantityAvailable": 99,
    "ProductId": 771,
    "id": "cfbb6333-ed9f-49e7-8640-bb920d5c9106",
    "Outbox": {
        "$type": "System.Collections.Generic.HashSet`1[[AdventureWorksCosmos.UI.Infrastructure.IDocumentMessage, AdventureWorksCosmos.UI]], System.Core",
        "$values": []
    },
    "Inbox": {
        "$type": "System.Collections.Generic.HashSet`1[[AdventureWorksCosmos.UI.Infrastructure.IDocumentMessage, AdventureWorksCosmos.UI]], System.Core",
        "$values": [
            {
                "$type": "AdventureWorksCosmos.UI.Models.Orders.ItemPurchased, AdventureWorksCosmos.UI",
                "ProductId": 771,
                "Quantity": 1,
                "Id": "2ab2108c-9698-49e8-93de-a3ced453836a"
            }
        ]
    },
    "_rid": "WQk4AKSQMwACAAAAAAAAAA==",
    "_self": "dbs/WQk4AA==/colls/WQk4AKSQMwA=/docs/WQk4AKSQMwACAAAAAAAAAA==/",
    "_etag": "\"060077c2-0000-0000-0000-5b71d8a10000\"",
    "_attachments": "attachments/",
    "_ts": 1534187681
}

We now have effective document messaging between our documents!

Well, almost.

In the next post, we'll walk through what to do when things go wrong: failures and retries.

Life Beyond Distributed Transactions: An Apostate's Implementation - Document Example

Jimmy Bogard — Thu, 09 Aug 2018 15:59:41 +0000

Posts in this series:

Sample code from this series

In the last post, I walked through the "happy path" scenario of coordinated communication/activities between multiple resources that otherwise can't participate in a transaction. In this post, I'll walk through a code example of building out document coordination in Azure Cosmos DB. My starting point is this set of code for approving an invoice and updating stock:

[HttpPost]
public async Task<IActionResult> Approve(Guid id)  
{
    var orderRequest = await _orderRepository.GetItemAsync(id);

    orderRequest.Approve();

    await _orderRepository.UpdateItemAsync(orderRequest);

    foreach (var lineItem in orderRequest.Items)
    {
        var stock = (await _stockRepository
            .GetItemsAsync(s => s.ProductId == lineItem.ProductId))
            .FirstOrDefault();

        stock.QuantityAvailable -= lineItem.Quantity;

        await _stockRepository.UpdateItemAsync(stock);
    }

    return RedirectToPage("/Orders/Show", new { id });
}

The repositories in my example are straight from the example code when you download a sample application in the Azure Portal, and just wrap the underlying DocumentClient.

Modeling our document

First, we need to baseline our document messages. These objects can be POCOs, but we still need some base information. Since we want to enforce idempotent actions, we need to be able to distinguish between different messages. The easiest way to do so is with a unique identifier per message:

public interface IDocumentMessage  
{
    Guid Id { get; }
}

Since our documents need to store and process messages in an inbox/outbox, we need to build out our base Document class to include these items. We can also build a completely separate object for our inbox/outbox, but for simplicity sake, we'll just use a base class:

public abstract class DocumentBase  
{
    [JsonProperty(PropertyName = "id")]
    public Guid Id { get; set; }

    private HashSet<IDocumentMessage> _outbox 
        = new HashSet<IDocumentMessage>(DocumentMessageEqualityComparer.Instance);
    private HashSet<IDocumentMessage> _inbox
        = new HashSet<IDocumentMessage>(DocumentMessageEqualityComparer.Instance);

    public IEnumerable<IDocumentMessage> Outbox
    {
        get => _outbox;
        protected set => _outbox = value == null
            ? new HashSet<IDocumentMessage>(DocumentMessageEqualityComparer.Instance)
            : new HashSet<IDocumentMessage>(value, DocumentMessageEqualityComparer.Instance);
    }

    public IEnumerable<IDocumentMessage> Inbox
    {
        get => _inbox;
        protected set => _inbox = value == null
            ? new HashSet<IDocumentMessage>(DocumentMessageEqualityComparer.Instance)
            : new HashSet<IDocumentMessage>(value, DocumentMessageEqualityComparer.Instance);
    }

Each of our mailboxes are a HashSet, to ensure we enforce uniqueness of document messages inside our document. We wrap our mailboxes in a couple of convenience properties for storage purposes (since our documents are serialized using JSON.NET, we have to model appropriately for its serialization needs).

We're using a custom equality comparer for document messages based on that interface and ID we added earlier:

public class DocumentMessageEqualityComparer  
    : IEqualityComparer<IDocumentMessage>
{
    public static readonly DocumentMessageEqualityComparer Instance
        = new DocumentMessageEqualityComparer();

    public bool Equals(IDocumentMessage x, IDocumentMessage y)
    {
        return x.Id == y.Id;
    }

    public int GetHashCode(IDocumentMessage obj)
    {
        return obj.Id.GetHashCode();
    }
}

With this, we can make sure that our document messages only exist in our inbox/outboxes once (assuming we can pick unique GUIDs).

Next, we need to be able to send a message in our DocumentBase class:

protected void Send(IDocumentMessage documentMessage)  
{
    if (_outbox == null)
        _outbox = new HashSet<IDocumentMessage>(DocumentMessageEqualityComparer.Instance);

    _outbox.Add(documentMessage);
}

We have to check that the outbox exists and create it if it's not (due to serialization, it might not exist), then simply add the document message to the outbox.

To process a document message, we need to make sure this action is idempotent. To check for idempotency, we'll examine our Inbox before executing the action. We can wrap this all up in a single method that our derived documents will use:

protected void Process<TDocumentMessage>(  
    TDocumentMessage documentMessage, 
    Action<TDocumentMessage> action)
    where TDocumentMessage : IDocumentMessage
{
    if (_inbox == null)
        _inbox = new HashSet<IDocumentMessage>(DocumentMessageEqualityComparer.Instance);

    if (_inbox.Contains(documentMessage))
        return;

    action(documentMessage);

    _inbox.Add(documentMessage);
}

Our derived documents will need to call this method to process their messages with the idempotency check. Once a message is processed successfully, we'll add it to the inbox. And since our transaction boundary is the document, if something fails, the action never happened and the message never gets stored to the inbox. Only by keeping our inbox, outbox, and business data inside a transaction boundary can we guarantee all either succeeds or fails.

Refactoring our action

Now that we have our basic mechanism of storing and processing messages, we can refactor our original action. It was split basically into two actions - one of approving the invoice, and another of updating stock.

We need to "send" a message from our Order to Stock. But what should that message look like? A few options:

Command, "update stock"
Event, "order approved"
Event, "item purchased"

If I go with a command, I'm coupling the primary action with the intended side effect. But what if this side effect needs to change? Be removed? I don't want burden the main Order logic with that.

What about the first event, "order approved"? I could go with this - but looking at the work done and communication, Stock doesn't care that an order was approved, it only really cares if an item was purchased. Approvals are really the internal business rules of an order, but the ultimate side effect is that items finally become "purchased" at this point in time. So if I used "order approved", I'd be coupling Stock to the internal business rules of Order. Even though it's an event, "order approved" concerns internal business processes that other documents/services shouldn't care about.

Finally, we have "item purchased". This most closely matches what Stock cares about, and removes any kind of process coupling between these two aggregates. If I went with the macro event, "order approved", I'd still have to translate that to what it means for Stock.

With this in mind, I'll create a document message representing this event:

public class ItemPurchased : IDocumentMessage  
{
    public int ProductId { get; set; }
    public int Quantity { get; set; }
    public Guid Id { get; set; }
}

I know how much of which product was purchased, and that's enough for Stock to deal with the consequences of that event.

My Order class then models its Approve method to include sending these new messages:

public void Approve()  
{
    Status = Status.Approved;
    foreach (var lineItem in Items)
    {
        Send(new ItemPurchased
        {
            ProductId = lineItem.ProductId,
            Quantity = lineItem.Quantity,
            Id = Guid.NewGuid()
        });
    }
}

I don't have an idempotency check here (if the order is already approved, do nothing), but you get the idea.

On the Stock side, I need to add a method to process the ItemPurchased message:

public void Handle(ItemPurchased message)  
{
    Process(message, e =>
    {
        QuantityAvailable -= e.Quantity;
    });
}

Finally, we need some way of linking our ItemPurchased message with the Stock, and that's the intent of an IDocumentMessageHandler:

public interface IDocumentMessageHandler<in T>  
    where T : IDocumentMessage
{
    Task Handle(T message);
}

The part of our action that loaded up each Stock is the code we'll put into our handler:

public class UpdateStockFromItemPurchasedHandler  
    : IDocumentMessageHandler<ItemPurchased>
{
    private readonly IDocumentDBRepository<Stock> _repository;

    public UpdateStockFromItemPurchasedHandler(
        IDocumentDBRepository<Stock> repository) 
        => _repository = repository;

    public async Task Handle(ItemPurchased message)
    {
        var stock = (await _repository
            .GetItemsAsync(s => s.ProductId == message.ProductId))
            .Single();

        stock.Handle(message);

        await _repository.UpdateItemAsync(stock);
    }
}

Not that exciting, as our document will handle the real business logic of handling the request. This class just connects the dots between an IDocumentMessageHandler and some DocumentBase instance.

With these basic building blocks, we'll modify our action to only update the Order instance:

[HttpPost]
public async Task<IActionResult> Approve(Guid id)  
{
    var orderRequest = await _orderRepository.GetItemAsync(id);

    orderRequest.Approve();

    await _orderRepository.UpdateItemAsync(orderRequest);

    return RedirectToPage("/Orders/Show", new { id });
}

Now when we approve our order, we only create messages in the outbox, which get persisted along with the order. If I look at the saved order in Cosmos DB, I can verify the items are persisted:

{
    "Items": [
        {
            "Quantity": 1,
            "ListPrice": 3399.99,
            "ProductId": 771,
            "ProductName": "Mountain-100 Silver, 38",
            "Subtotal": 3399.99
        }
    ],
    "Status": 2,
    "Total": 3399.99,
    "Customer": {
        "FirstName": "Jane",
        "LastName": "Doe",
        "MiddleName": "Mary"
    },
    "id": "8bf4bda2-3796-431e-9936-8511243352d2",
    "Outbox": {
        "$type": "System.Collections.Generic.HashSet`1[[AdventureWorksCosmos.UI.Infrastructure.IDocumentMessage, AdventureWorksCosmos.UI]], System.Core",
        "$values": [
            {
                "$type": "AdventureWorksCosmos.UI.Models.Orders.ItemPurchased, AdventureWorksCosmos.UI",
                "ProductId": 771,
                "Quantity": 1,
                "Id": "987ce801-e7cf-4abf-aba7-83d7eed00610"
            }
        ]
    },
    "Inbox": {
        "$type": "System.Collections.Generic.HashSet`1[[AdventureWorksCosmos.UI.Infrastructure.IDocumentMessage, AdventureWorksCosmos.UI]], System.Core",
        "$values": []
    },
    "_rid": "lJFnANVMlwADAAAAAAAAAA==",
    "_self": "dbs/lJFnAA==/colls/lJFnANVMlwA=/docs/lJFnANVMlwADAAAAAAAAAA==/",
    "_etag": "\"02002652-0000-0000-0000-5b48f2140000\"",
    "_attachments": "attachments/",
    "_ts": 1531507220
}

In order to get that polymorphic behavior for my IDocumentMessage collections, I needed to configure the JSON serializer settings in my repository:

_client = new DocumentClient(new Uri(Endpoint), Key,  
    new JsonSerializerSettings
    {
        TypeNameHandling = TypeNameHandling.Auto
    });

With these pieces in place, I've removed the process coupling between updating an order's status and updating stock items using document messaging. Of course, we don't actually have anything dispatching our messages. We'll cover the infrastructure for dispatching messages in the next post.

Life Beyond Distributed Transactions: An Apostate's Implementation - Document Coordination

Jimmy Bogard — Tue, 07 Aug 2018 18:24:19 +0000

Posts in this series:

Sample code from this series

Quick note - I've updated this post to use the more accurate term "Document" than the less-accurate, DDD-specific term "Aggregate". In the images, it still has the older term "Aggregate", let's pretend I fixed all those images.

In the last post, I walked through the general problem of distributed transactions, and some potential ideas around coordinating activities between similar or disparate resources. In a lot of cases, I would prefer to simply wrap our actions in a transaction, but as the resources I'm trying to transact together move further apart (process-wise or network-wise), transactions between these multiple resources becomes impossible.

If I can only perform reliable transactions with a single resource at a time, whether that resource is a database or single record in Cosmos DB, I need to design my interactions so that both the business data and communications transact together. That means that my communication needs to be in the same transactional store as my business data!

In the case of Azure Cosmos DB, I need to place my communication that I both send and receive inside my document. I store incoming communication because any request to change data needs to be idempotent. I could just have some business rule "all actions must be idempotent", I can also achieve this by simply storing the incoming requests and checking to see if I've already processed this request before performing the action.

For outgoing communication, I can't interact with any other resource transactionally, so I store outgoing communication on what I can control - the document. So our final document looks like:

When I perform an operation, communication and business data either save successfully or the entire operation rolls back.

Dispatching Changes

When I find I have an operation that needs to affect more than one resource, I then use my outbox to communicate with those external resources:

I can then safely transact that single resource:

Now that I've completed the first action, I can now begin my interaction with other resources. I use a dispatcher to do so - something that's responsible for reading from a resource's outbox:

And pass them to the correct resource:

Each resource receives the incoming message, and performs 2 actions:

Affect the business data
Store the incoming message in the inbox

This entire operation is transactional:

With that resource successful, the dispatcher moves on to the next resource. However, this operation can fail:

When this happens, we move the message to an external retry queue:

At this point the dispatcher is complete - but because we still have a failed resource, we can't remove our event from the original document's outbox.

Retries

In our half-completed state, we need to re-process the outbox for our document. So our dispatcher wakes up from the Retry message and reads the messages in the original document's outbox:

The dispatcher dispatches our message again to each of the receiving resources, processing each one in turn. Our first resource has already processed this message, which it knows because it's checked its Inbox before changing the data (making it idempotent):

With that complete, the dispatcher can move on to the (previously) failing document:

And that document can now succeed. Of course, there might be a bug or something actively preventing us from succeeding - so in that case we'll need to design explicit failure strategies. For now, let's assume that our operations will eventually succeed. Once we've confirmed that we've successfully dispatched our message to all receivers, we can go back to our original document and remove the message:

This is then saved in its own individual transaction:

In each step of the way, we have to work in smaller transactional steps that can be individually retried at least once, since each operation is idempotent. By storing our communication in the same transactional boundary of our business data, we can introduce reliable resource coordination when each of our resources isn't able to participate in that transaction.

In the next post, I'll walk through some code examples of this pattern in Cosmos DB.

Life Beyond Distributed Transactions: An Apostate's Implementation - A Primer

Jimmy Bogard — Wed, 01 Aug 2018 13:56:53 +0000

Posts in this series:

Sample code from this series

For those working with SQL databases, working with transactions is more or less a given. The most we may need to worry about is:

Using the appropriate isolation level
Not doing too much in a single transaction to prevent excessive locks

The vast majority of applications I see can blissfully ignore the inner workings of transactions in database. We take for granted that our operations on one or more rows are either committed or rolled back.

Things get more complicated when we start dealing with transactions no longer confined to a single resource, or in the case of many NoSQL databases, multiple entities. In many NoSQL databases, transactions are limited to a single entity/record. And if multi-entity transaction are supported, there are a number of limitations that might make that choice undesirable.

When we have two non-transactional resources, we know we have a number of overall patterns to try to coordinate these actions (covered in my Refactoring Towards Resilience series:

All of these options assume "I must have these two actions temporally coupled" and have them happen at the same time.

But what if that wasn't the case? What if we moved away from trying to coordinate two actions, and had more loose coupling between our resources? What might that look like?

And that's the main scope of Pat Helland's paper, Life Beyond Transactions: An Apostate's Opinion. In this paper, Pat describes a mechanism to overcome the fundamental issue of coordinating actions between resources when our transactions only cover a single entity - messaging!

We use some sort of messaging where our messages are saved inside the entities to direct to other entities:

Since the scope of a transaction is a single entity, if we need to affect other entities, we can't do that directly. Instead, we store the intent to affect change as a message inside our entity. The transaction covers our business data, and communication to the outside world.

This certainly isn't the only way to tackle this problem, as I could use the Saga pattern as a means to manage failures between multi-entity activities. Caitie McCaffrey has a great talk about this, but for my situation, I couldn't directly use the Saga pattern, since that implied there was some sort of logical "undo". Instead, I wanted to be able to any of the many coordination patterns available to me, including the Saga pattern.

Before we get into implementation details, let's look at some real code in a real database that doesn't completely allow multi-entity transactions.

Real World Example

Let's suppose I have an ecommerce application, where I can view products, add items to a cart, check out, then finally, approve orders. As part of approving an order, I need to decrement stock. We'll skip any complex business rules, like negative stock, reservations, and the like. Just to keep things simple, when we order, we just subtract the quantity order from our stock reserve:

var orderRequest = await _orderRepository.GetItemAsync(id);

orderRequest.Approve();

await _orderRepository.UpdateItemAsync(orderRequest);

foreach (var lineItem in orderRequest.Items)  
{
    var stock = (await _stockRepository
        .GetItemsAsync(s => s.ProductId == lineItem.ProductId))
        .Single();

    stock.QuantityAvailable -= lineItem.Quantity;

    await _stockRepository.UpdateItemAsync(stock);
}

In my case, I'm using Azure Cosmos DB, which supports a variety of consistency levels. Azure Cosmos DB also supports multi-document transactions, but only in the form of stored procedures and functions, and even then, only with some limitations. Once you introduce other resources into the mix, like Azure Service Bus, Azure SQL Database, or really anything outside a single partition key, we can no longer use transactions.

With this in mind, we look back at our original code, and ask ourselves, "Do this action need to couple these resources, or can we decouple them?" There are certainly cases where we need to coordinate two actions (using a coordinator as we saw before), but there are many cases we don't, and don't want to incur the cost of a larger-scoped transaction.

For our above case, do we need to deduct stock right at the time we approve an order? Or can it happen later? According to the business, deducting stock doesn't need to happen immediately, but it does need to happen, eventually.

In the next few posts, I'll walk through building out a mechanism for communication with other entities (and even other resources), to see how we might build out an atomic communication system with Azure Cosmos DB as our example.