Using the Transactional Outbox Pattern with Azure Cosmos DB for Guaranteed Delivery of Domain Events
Tags: azure, cosmosdb, dotnet, ddd
Published: 2021-09-22 11:27:04

---
title: Using the Transactional Outbox Pattern with Azure Cosmos DB for Guaranteed Delivery of Domain Events
published: true
date: 2021-09-22 10:08:55 UTC
tags: azure,cosmosdb,dotnet,ddd
canonical_url: http://partlycloudy.blog/2021/09/22/using-the-transactional-outbox-pattern-with-azure-cosmos-db-for-guaranteed-delivery-of-domain-events/
cover_image: https://heartofcode.files.wordpress.com/2021/08/pawel-czerwinski-u78zbap7rym-unsplash.jpg
---

_This article is about how to build a resilient architecture for distributed applications leveraging basic Domain-Driven Design and the Transactional Outbox Pattern with Azure Cosmos DB and Azure Service Bus._

(**STOP!** This article has been migrated to (and will be kept up-to-date in) the Azure Architecture Center: <https://docs.microsoft.com/en-us/azure/architecture/best-practices/transactional-outbox-cosmos>)

Microservice architectures become increasingly popular these days and promise to solve problems like scalability, maintainability, agility etc. by splitting your once monolithic application into smaller (micro-) services that can operate and scale on their own. While you want these services to be independent from other services of your application (no tight coupling!), this also means managing data necessary to operate independently within a dedicated datastore per service. In such a distributed, microservice-oriented application, you will end up running multiple datastores where data is often replicated between different services of your application.

How does such a service get hold of all the data it needs to run properly? Typically, you use a messaging solution like RabbitMQ, Kafka or Azure Service Bus that distributes events from your microservices via a messaging bus after a business object has been created, modified, deleted etc. By following such an approach, you avoid calling and therefor tightly coupling your services together.

* * *

Let’s have a look at the “Hello, World!” example in that space: in an _Ordering_ service, when a user wants to create a new order, the service would receive the payload from a client application via a REST endpoint, it would map the data to the internal representation of an _Order_ object (validate the data) and after a successful commit to the database, it would publish an _OrderCreated_ event to a message bus. Any other service interested in newly created orders (e.g. an _Inventory_ or _Invoicing_ service), would subscribe to these _OrderCreated_ messages and process them accordingly. The following pseudo code shows, how this would typically look like in the _Ordering_ service:

```csharp

CreateNewOrder(CreateOrderDto order){
  // validate the incoming data
  ...
  // do some other stuff
  ...
  // Save the object to the database
  var result = _orderRespository.Create(order);
  // finally, publish the respective event
  _messagingService.Publish(new OrderCreatedEvent(result));
  return Ok();
}

```

That all sounds wonderful until something unexpected happens between the highlighted lines of the implementation. Imagine you successfully saved the _Order_ object to the database, are in the middle of publishing the event to the message bus and your service crashes, the message bus is not available (due to whatever reason) or the network between your service and the messaging system has a problem?!

![](https://heartofcode.files.wordpress.com/2021/08/image-1.png?w=1024)

In a nutshell: you cannot publish the _OrderCreated_ event to “the outside world” although the actual object has already been saved. Now what? You end up having a severe problem and start saving the events to a secondary service like an Azure Storage Account to process them later or – even worse – writing code that keeps track of all the events that haven’t already been published in memory. Worst case, you end up having data inconsistencies in your other services, because events are lost. Debugging these kinds of problems is pure nightmare and you want to avoid these situations whatever it takes.

## Solution

So, what is the solution for the problem described above? There is a well-known pattern called **Transactional Outbox Pattern** that will help us here. The pattern ensures that events will be saved in a datastore (typically in an _Outbox_ table of your database) before ultimately pushing them to the message broker. If the business object and the corresponding events are saved **within the same database transaction** , it is also guaranteed that no data will be lost – either everything will be committed or rolled back. To eventually publish the event, a different service or worker process can query the _Outbox_ table for new entries, publish the events and mark them as “processed”. Using this pattern, you ensure that events will never be lost after creating or modifying a business object.

![](https://heartofcode.files.wordpress.com/2021/09/image.png?w=1024)

In a traditional database, the implementation in your service is fairly easy. If you e.g. use Entity Framework Core, you will use your EF context to create a database transaction, save the business object and the event and commit the transaction – or do a rollback in case of an error. Even the worker service that is processing the events data is straightforward: periodically query the _Outbox_ table for new entries, publish the event payload to the message bus and mark the entry as “processed”.

Of course, in real-life things are not as easy as they might look like in the first place. Most importantly, you need to make sure that the order of the events, as they happened in the application, is preserved so that an _OrderUpdated_ event doesn’t get published before an _OrderCreated_ event.

## How to implement the pattern with Azure Cosmos DB?

So far, we discussed the **Transactional Outbox Pattern** in theory and how it can help to implement reliable messaging in distributed applications. But how can we use the pattern in combination with Azure Cosmos DB and leverage some of the unique features like Cosmos DB Change Feed to make our developer life easier? Let’s have a detailed look at it by implementing a sample service that is responsible for managing _Contact_ objects (Firstname, Lastname, Email, Company Information etc.). The sample service uses the CQRS pattern and follows basic Domain-Driven Design concepts.

### Azure Cosmos DB

![Cosmos DB Logo](https://heartofcode.files.wordpress.com/2021/09/cosmosdb-logo-1.png)

First, Azure Cosmos DB – in case you haven’t heard of the service so far – is a globally distributed NoSQL database service that lets you use different APIs like Gremlin, MongoDB and the Core SQL API (which we will focus on in this article) to manage your application data – with 99.999% SLAs on availability. Cosmos DB can automatically horizontally scale your database, promising unlimited storage and throughput. The service keeps its promise by partitioning your data based on a user-defined key provided at container (the place where your documents will be stored) creation, the so-called “partition key”. That means your data is divided into logical subsets within a container based on that partition key – you may have also heard of the term “sharding” in case you are familiar with MongoDB. Under the hood, Cosmos DB will then distribute these logical subsets of your data across (multiple) physical partitions.

### Cosmos DB Transactional Batches

Due to the mechanism described above, transactions in Cosmos DB work slightly different than you might be familiar with in relational database systems. Cosmos DB transactions – called “transactional batches” – operate on a single logical partition and therefor guarantee ACID properties. That means, you can’t save two documents in a transactional batch operation in different containers or even logical partitions (read: with different “partition keys”). Due to that mechanism, the implementation of the **Transactional Outbox Pattern** with Cosmos DB is slightly different compared to the approach in a relational database where you would simply have two tables: one for the business object and one _Outbox_ table for the events. In Cosmos DB, we need to save the business object and the events data in the same logical partition (in the same container). The following chapters describe, how this can be achieved.

### Custom Container Context, Repositories and UnitOfWork

The most important part of the implementation is a “custom container context” that is responsible of keeping track of objects that need to be saved in the same transactional batch. Think of a very lightweight “Entity Framework Context” maintaining a list of created/modified objects that also knows in which Cosmos DB container (and logical partition) these objects need to be saved. Here’s the interface for it:

```csharp

public interface IContainerContext
{
    public Container Container { get; }
    public List<IDataObject<Entity>> DataObjects { get; }
    public void Add(IDataObject<Entity> entity);
    public Task<List<IDataObject<Entity>>> 
      SaveChangesAsync(CancellationToken cancellationToken = default);
    public void Reset();
}

```

The list within the “container context” component will hold _Contact_ as well as _DomainEvent_ objects and both will be put in the same container – yes, we are mixing multiple types of objects in the same Cosmos DB container and use a _Type_ property to distinct between an “entity” and an “event”.

For each type there exists a dedicated repository that defines/implements the data access. The _Contact_ repository interface offers the following methods:

```csharp

public interface IContactsRepository
{
    public void Create(Contact contact);
    public Task<(Contact, string)> ReadAsync(Guid id, string etag);
    public Task DeleteAsync(Guid id, string etag);
    public Task<(List<(Contact, string)>, bool, string)> 
      ReadAllAsync(int pageSize, string continuationToken);
    public void Update(Contact contact, string etag);
}

```

The _Event_ repository looks similar, except that there is only one method to create new events in the store:

```csharp

public interface IEventRepository
{
    public void Create(ContactDomainEvent e);
}

```

The implementations of both repository interfaces get a reference via dependency injection to an _IContainerContext_ instance to make sure that both **operate on the same context**.

The final component is a _UnitOfWork_ that is responsible for committing the changes held in the _IContainerContext_ instance to the database:

```csharp

public class UnitOfWork : IUnitOfWork
{
    private readonly IContainerContext _context;
    public IContactRepository ContactsRepo { get; }

    public UnitOfWork(IContainerContext ctx, IContactRepository cRepo)
    {
        _context = ctx;
        ContactsRepo = cRepo;
    }

    public Task<List<IDataObject<Entity>>> 
        CommitAsync(CancellationToken cancellationToken = default)
    {
        return _context.SaveChangesAsync(cancellationToken);
    }
}

```csharp

We now have the components in place for data access. Let’s see, how events are created and published.

### Event Handling – Creation and Publication

Every time a _Contact_ object is created, modified or (soft-) deleted, we want the system to raise a corresponding event, so that we can notify other services interested in these changes immediately. Core of the solution provided in this article is a combination of Domain-Driven Design and making use of the mediator pattern as proposed by Jimmy Bogard [[1](https://lostechies.com/jimmybogard/2014/05/13/a-better-domain-events-pattern/)]. He suggests maintaining a list of domain events that happened due to modifications of the domain object and “publish” these events ultimately **before saving the actual object to the database**. The list of changes is kept in the domain object itself, so that no other component can modify the chain of events. The behavior of maintaining such events (_IEvent_ instances) in a domain object is defined via an interface _IEventEmitter<IEvent>_ and implemented in an abstract _DomainEntity_ class:

```csharp

public abstract class DomainEntity : Entity, IEventEmitter<IEvent>
{
...
...
    private readonly List<IEvent> _events = new();

    [JsonIgnore] public IReadOnlyList<IEvent> DomainEvents => 
        _events.AsReadOnly();

    public virtual void AddEvent(IEvent domainEvent)
    {
        var i = _events.FindIndex(0, e => e.Action == domainEvent.Action);
        if (i < 0)
        {
            _events.Add(domainEvent);
        }
        else
        {
            _events.RemoveAt(i);
            _events.Insert(i, domainEvent);
        }
    }
...
...
}

```

When it comes to raising/adding domain events, it’s the responsibility of the _Contact_ object. As mentioned before, the _Contact_ entity follows basic DDD concepts. In this case, it means that you can’t modify any domain properties “from outside”: you don’t have any public setters in the class. Instead, it offers dedicated methods to manipulate the state and is therefore able to raise the appropriate events for a certain modification (e.g. “NameUpdated”, “EmailUpdated” etc.).

Here’s an example when updating the name of a contact – the event is raised at the highlighted row:

```csharp

public void SetName(string firstName, string lastName)
{
    if (string.IsNullOrWhiteSpace(firstName) || 
        string.IsNullOrWhiteSpace(lastName))
    {
        throw 
            new ArgumentException("FirstName or LastName may not be empty");
    }

    Name = new Name(firstName, lastName);

    if (IsNew) return;

    AddEvent(new ContactNameUpdatedEvent(Id, Name));
    ModifiedAt = DateTimeOffset.Now;
}

```

The corresponding _ContactNameUpdatedEvent_ that simply keeps track of the individual changes to the domain object looks as follows:

```csharp

public class ContactNameUpdatedEvent : ContactDomainEvent
{
    public Name Name { get; }

    public ContactNameUpdatedEvent(Guid contactId, Name contactName) : 
        base(Guid.NewGuid(), contactId, nameof(ContactNameUpdatedEvent))
    {
        Name = contactName;
    }
}

```

So far, we only “log” the events to the domain object and nothing gets published or even saved to the database. How can we persist the events in combination with the business object? Well, it’s done right before saving the domain object to Cosmos DB.

Within the _SaveChangesAsync_ method of the _IContainerContext_ implementation, we simply loop over all objects tracked in the container context and publish the events maintained in these objects. It’s all done in a private _RaiseDomainEvents_ method (_dObjs_ is the list of tracked entities of the container context):

```csharp

private void RaiseDomainEvents(List<IDataObject<Entity>> dObjs)
{
    var eventEmitters = new List<IEventEmitter<IEvent>>();

    // Get all EventEmitters
    foreach (var o in dObjs)
        if (o.Data is IEventEmitter<IEvent> ee)
            eventEmitters.Add(ee);

    // Raise Events
    if (eventEmitters.Count <= 0) return;
    foreach (var evt in eventEmitters
        .SelectMany(eventEmitter => eventEmitter.DomainEvents))
        _mediator.Publish(evt);
}

```

On line 14 (highlighted), we use the _MediatR_ library to publish an event within our application – this is possible, because all events – like _ContactNameUpdatedEvent_ – also implement the _INotification_ interface of the _MediatR_ package.

Of course, we also need to handle these events – and this is where the _IEventsRepository_ implementation comes into play. Let’s have a look at such an event handler:

```csharp

public class ContactNameUpdatedHandler :
    INotificationHandler<ContactNameUpdatedEvent>
{
    private IEventRepository EventRepository { get; }

    public ContactNameUpdatedHandler(IEventRepository eventRepo)
    {
        EventRepository = eventRepo;
    }

    public Task Handle(ContactNameUpdatedEvent notification, 
        CancellationToken cancellationToken)
    {
        EventRepository.Create(notification);
        return Task.CompletedTask;
    }
}

```

You can see, that an _IEventRepository_ instance gets injected in the handler class via the constructor. So, as soon as an _ContactNameUpdatedEvent_ gets published in the application, the _Handle_ method gets invoked and uses the events repository instance to create a notification object – that ultimately lands in the list of tracked objects in the _IContainerContext_ object and therefore **will become part of the objects that will be saved in the same transactional batch** to Cosmos DB.

Here are the important parts of the implementation of _IContainerContext_:

```csharp

private async Task<List<IDataObject<Entity>>> 
    SaveInTransactionalBatchAsync(List<IDataObject<Entity>> dObjs,
        CancellationToken cancellationToken)
{
    if (dObjs.Count > 0)
    {
        var pk = new PartitionKey(dObjs[0].PartitionKey);
        var tb = Container.CreateTransactionalBatch(pk);
        dObjs.ForEach(o =>
        {
            TransactionalBatchItemRequestOptions tro = null;

            if (!string.IsNullOrWhiteSpace(o.Etag))
                tro = new TransactionalBatchItemRequestOptions 
                { 
                    IfMatchEtag = o.Etag 
                };

            switch (o.State)
            {
                case EntityState.Created:
                    tb.CreateItem(o);
                    break;
                case EntityState.Updated or EntityState.Deleted:
                    tb.ReplaceItem(o.Id, o, tro);
                    break;
            }
        });

        var tbResult = await tb.ExecuteAsync(cancellationToken);
...
[check for return codes etc.]
...
    }

    var result = new List<IDataObject<Entity>>(dObjs);
    // reset internal list
    DataObjects.Clear();
    return result;
}

```

### Let Cosmos DB Shine

We now have everything in place: any time a domain object gets created or modified, it adds a corresponding domain event, which will be put in the same container context (via a separate repository) for object tracking and finally saved – both the domain object and the events – in the same transactional batch to Cosmos DB.

This is how the process works so far (e.g. for updating the name on a contact object):

1. _SetName_ is invoked on the contact object
2. Event _ContactNameUpdated_ is added to the list of events in the **domain object**
3. ContactRepository _Update_ method is invoked which adds the domain object to the container context. **Object is now “tracked”**.
4. _CommitAsync_ is invoked on the UnitOfWork object which in turn calls _SaveChangesAsync_ on the container context
5. Within _SaveChangesAsync_, all events in the list of the domain object get published by a _MediatR_ instance and are added via the EventsRepository to the **same container context**
6. After that, in _SaveChangesAsync_, a _TransactionalBatch_ is created which will hold both the contact object and the event
7. The _TransactionalBatch_ is executed and the data is committed to Cosmos DB
8. _SaveChangesAsync_ and _CommitAsync_ successfully return
9. End of the update process

* * *

Let’s have a closer look now at how each type gets persisted to a container. In the code snippets above, you saw that objects saved to Cosmos DB will be wrapped in a _DataObject_ instance. Such an object provides common properties like _Id_, _PartitionKey_, _Type_, _State_ (like e.g. _Created_, _Updated_ etc. – won’t be persisted in Cosmos DB), _Etag_ (for optimistic locking [[2](https://docs.microsoft.com/en-us/azure/cosmos-db/sql/database-transactions-optimistic-concurrency#optimistic-concurrency-control)]), _TTL_ (Time-To-Live property for automatic cleanup of old documents [[3](https://docs.microsoft.com/en-us/azure/cosmos-db/sql/time-to-live)]) and – of course – the _Data_ itself. All of this is defined in a generic interface called _IDataObject_ and used by the repositories and the container context:

```csharp

public interface IDataObject<out T> where T : Entity
{
    string Id { get; }
    string PartitionKey { get; }
    string Type { get; }
    T Data { get; }
    string Etag { get; set; }
    int Ttl { get; }
    EntityState State { get; set; }
}

```

Objects wrapped in a _DataObject_ instance and saved to the database will then look like this (_Contact_ and _ContactNameUpdatedEvent_):

```json

// Contact document/object - after creation

{
    "id": "b5e2e7aa-4982-4735-9422-c39a7c4af5c2",
    "partitionKey": "b5e2e7aa-4982-4735-9422-c39a7c4af5c2",
    "type": "contact",
    "data": {
        "name": {
            "firstName": "Bertram",
            "lastName": "Gilfoyle"
        },
        "description": "This is a contact",
        "email": "bg@piedpiper.com",
        "company": {
            "companyName": "Pied Piper",
            "street": "Street",
            "houseNumber": "1a",
            "postalCode": "092821",
            "city": "Palo Alto",
            "country": "US"
        },
        "createdAt": "2021-09-22T11:07:37.3022907+02:00",
        "deleted": false,
        "id": "b5e2e7aa-4982-4735-9422-c39a7c4af5c2"
    },
    "ttl": -1,
    "_etag": "\"180014cc-0000-1500-0000-614455330000\"",
    "_ts": 1631868211
}

// after setting a new name - this is how an event document looks like

{
    "id": "d6a5f4b2-84c3-4ac7-ae22-6f4025ba9ca0",
    "partitionKey": "b5e2e7aa-4982-4735-9422-c39a7c4af5c2",
    "type": "domainEvent",
    "data": {
        "name": {
            "firstName": "Dinesh",
            "lastName": "Chugtai"
        },
        "contactId": "b5e2e7aa-4982-4735-9422-c39a7c4af5c2",
        "action": "ContactNameUpdatedEvent",
        "id": "d6a5f4b2-84c3-4ac7-ae22-6f4025ba9ca0",
        "createdAt": "2021-09-17T10:50:01.1692448+02:00"
    },
    "ttl": 120,
    "_etag": "\"18005bce-0000-1500-0000-614456b80000\"",
    "_ts": 1631868600
}

```

You see that the _Contact_ and _ContactNameUpdatedEvent_ (type: _domainEvent_) documents have the same partition key – hence both documents will be persisted in the same logical partition. And, if you want to keep a log of changes of a contact object, you can also keep all the events in the container for good. Frankly, this is not what we want to have here in this sample. The purpose of persisting both object types is to make sure that **a)** events never get lost and **b)** they eventually get published to “the outside world” (e.g. to an Azure Service Bus topic).

* * *

So, to finally read the stream of events and send them to a message broker, let’s use one of the “unsung hero features” [[4](https://devblogs.microsoft.com/cosmosdb/change-feed-unsung-hero-of-azure-cosmos-db/)] of Cosmos DB: **Change Feed**.

The Change Feed is a persistent log of changes in your container that is operating in the background keeping track of modifications in the order they occurred – per logical partition. So, when you read the Change Feed, it is guaranteed that – for a certain partition key – you read the changes of those documents always in the correct order. This is mandatory for our scenario. Frankly, this is the reason why we put both the contact and corresponding event documents in the same partition: when we read the Change Feed, we can be sure that we never get an “updated” before a “created” event.

So, how do we read the Change Feed then? You have multiple options. The most convenient way is to use an Azure Function with a Cosmos DB trigger. Here, you have everything in place and if you want to host that part of the application on the Azure Functions service, you are good to go. Another option is to use the Change Feed Processor library [[5](https://docs.microsoft.com/en-us/azure/cosmos-db/sql/change-feed-processor)]. It lets you integrate Change Feed processing in your Web API e.g. as a background service (via _IHostedService_ interface). In this sample, we simply create a console application that uses the abstract class _BackgroundService_ for implementing long running background tasks in .NET Core.

Now, to receive the changes from the Cosmos DB Change Feed, you need to instantiate a _ChangeFeedProcessor_ object, register a handler method for message processing and start listening for changes:

```csharp

private async Task<ChangeFeedProcessor> 
    StartChangeFeedProcessorAsync()
{
    var changeFeedProcessor = _container
        .GetChangeFeedProcessorBuilder<ExpandoObject>(
            _configuration.GetSection("Cosmos")["ProcessorName"],
            HandleChangesAsync)
        .WithInstanceName(Environment.MachineName)
        .WithLeaseContainer(_leaseContainer)
        .WithMaxItems(25)
        .WithStartTime(new DateTime(2000, 1, 1, 0, 0, 0, DateTimeKind.Utc))
        .WithPollInterval(TimeSpan.FromSeconds(3))
        .Build();

    _logger.LogInformation("Starting Change Feed Processor...");
    await changeFeedProcessor.StartAsync();
    _logger.LogInformation("Change Feed Processor started.  
        Waiting for new messages to arrive.");
    return changeFeedProcessor;
}

```

The handler method is then responsible for processing the messages. In this sample, we keep things simple and publish the events to an Azure Service Bus topic which is partitioned for scalability and has the “deduplication” feature enabled [[6](https://docs.microsoft.com/en-us/azure/service-bus-messaging/duplicate-detection)] – more on that in a second. From there on, any service interested in changes to _Contact_ objects can subscribe to that topic and receive and process those changes for its own context.

You’ll also see that the Service Bus messages have a _SessionId_ property. By using sessions in Azure Service Bus, we guarantee that the order of the messages is preserved (FIFO) [[7](https://docs.microsoft.com/en-us/azure/service-bus-messaging/message-sessions)] – which is necessary for our use case. Here is the snippet that handles messages from the Change Feed:

```csharp

private async Task HandleChangesAsync(IReadOnlyCollection<ExpandoObject> changes, CancellationToken cancellationToken)
{
    _logger.LogInformation($"Received {changes.Count} document(s).");
    var eventsCount = 0;

    Dictionary<string, List<ServiceBusMessage>> partitionedMessages = new();

    foreach (var document in changes as dynamic)
    {
        if (!((IDictionary<string, object>)document).ContainsKey("type") ||
            !((IDictionary<string, object>)document).ContainsKey("data")) continue; // unknown doc type

        if (document.type == EVENT_TYPE) // domainEvent
        {
            string json = JsonConvert.SerializeObject(document.data);
            var sbMessage = new ServiceBusMessage(json)
            {
                ContentType = "application/json",
                Subject = document.data.action,
                MessageId = document.id,
                PartitionKey = document.partitionKey,
                SessionId = document.partitionKey
            };

            // Create message batch per partitionKey
            if (partitionedMessages.ContainsKey(document.partitionKey))
            {
                partitionedMessages[sbMessage.PartitionKey].Add(sbMessage);
            }
            else
            {
                partitionedMessages[sbMessage.PartitionKey] = new List<ServiceBusMessage> { sbMessage };
            }

            eventsCount++;
        }
    }

    if (partitionedMessages.Count > 0)
    {
        _logger.LogInformation($"Processing {eventsCount} event(s) in {partitionedMessages.Count} partition(s).");

        // Loop over each partition
        foreach (var partition in partitionedMessages)
        {
            // Create batch for partition
            using var messageBatch =
                await _topicSender.CreateMessageBatchAsync(cancellationToken);
            foreach (var msg in partition.Value)
                if (!messageBatch.TryAddMessage(msg))
                    throw new Exception();

            _logger.LogInformation(
                $"Sending {messageBatch.Count} event(s) to Service Bus. PartitionId: {partition.Key}");

            try
            {
                await _topicSender.SendMessagesAsync(messageBatch, cancellationToken);
            }
            catch (Exception e)
            {
                _logger.LogError(e.Message);
                throw;
            }
        }
    }
    else
    {
        _logger.LogInformation("No event documents in change feed batch. Waiting for new messages to arrive.");
    }
}

```

### What happens when errors occur?

In case of an error while processing the changes, the Change Feed library will restart reading messages at the position where it successfully processed the last batch. That means, if we already have processed 10.000 messages, are now in the middle of working on messages 10.001 to 10.0025 (we read documents in batches of 25 – see the initialization of the _ChangeFeedProcessor_ object) and an error happens, we can simply restart our process and it will pick up its work at position 10.001. The library keeps track of what has already been processed via a _Leases_ container.

Of course, it can happen that – out of the 25 messages that are then reprocessed – our application already sent 10 to Azure Service Bus. This is where the deduplication feature will save our life. Azure Service Bus will check if a message has already been added to a topic during a specified time window based on the application-controlled _MessageId_ property of a message. That property is set to the _Id_ of the event document, meaning Azure Service Bus will ignore and drop a message, if a certain event has already been successfully added to our topic.

In a typical “Transactional Outbox Pattern” implementation, we would now update the processed events and set a _Processed_ property to `true`, indicating that we successfully published it – and every now and then, the events would then be deleted to keep only the most recent records/documents. Of course, this could be addressed manually in the handler method.

**But we are using Cosmos DB!** We are already keeping track of events that were processed by using the Change Feed (in combination with the _Leases_ container). And to periodically clean-up the events, we can leverage another useful feature of Cosmos DB: Time-To-Live (_TTL_) on documents [3]. Cosmos DB provides the ability to automatically delete documents based on a _TTL_ property that can be added to a document – a timespan in seconds. The service will constantly check your container for documents with such a _TTL_ property and as soon as it has expired, Cosmos DB will remove it from the container (btw, using the remaining Request Units in a background task – it will never “eat up” RUs that you need to handle user requests).

* * *

So, when all the services work as expected, events will be processed and published fast – meaning within seconds. If we have an error in Cosmos DB, we do not even publish any events, because nothing (both business object as well as corresponding events) can’t be saved at all – users will receive an error.

The only thing we need to consider and therefore set an appropriate _TTL_ value on our _DomainEvent_ documents, is when the background worker application (Change Feed Processor) or the Azure Service Bus aren’t available.

How much time do we grant both components to eventually publish the events? Let’s have a look at the SLA for Azure Service Bus: Microsoft guarantees 99.9% uptime. That means a downtime of ~9h per year or ~44mins per month. Of course, we should give our service more time than a duration of a “possible outage” to process changes. Furthermore, we should also consider that our Change Feed Processor worker service can also be down. To be on the safe side, I’d pick 10 days (_TTL_ expects seconds, so a value of _864.000_) for a production environment. All components involved will then have more than one week to process / publish changes within our application. That should be enough even in the case of a “disaster” like an Azure region outage. And after 10 days, the container holding the events will be cleaned-up automatically by Cosmos DB.

**Sidenote:** As mentioned before, that’s not necessary. If you want to keep a log of all the changes for a _Contact_ object, you can also set the _TTL_ to _-1_ on event documents and Cosmos DB won’t purge them at all.

## Advantages

As a result, we now have a service that can handle _Contact_ business objects and every time such a contact gets added or modified, it will raise a corresponding event which will be published within the application right before saving the actual object. These events will be picked up by an event handler in our service and added to the same “database context” which will save both the business object and events in the same transaction to Cosmos DB. Hence, we can guarantee that no event will ever be lost. Furthermore, we leverage the Cosmos DB Change Feed then to publish the tracked events in the background to an Azure Service Bus Topic via a background worker service that makes use of the _ChangeFeedProcessor_ library.

So, in the end this is not a “traditional” implementation of the “Transactional Outbox Pattern”, because we leverage some features of Cosmos DB that makes things easier for a developer.

What are the advantages of this solution then? We have…

- guaranteed delivery of events
- guaranteed ordering of events and deduplication via Azure Service Bus
- no need to maintain an extra _Processed_ property that indicates a successful processing of an event document – the solution makes use of Time-To-Live and Change Feed features in Cosmos DB
- error proof processing of messages via ChangeFeedProcessor (or Azure Function)
- optional: you can add multiple Change Feed processors – each one maintaining its own “pointer” enabling additional scenarios
- if you use a multi-master/multi-region Cosmos DB account, you’ll even get “five nines” (99.999%) availability – which is outstanding!

## Summary

This sample demonstrated how to implement the **Transactional Outbox Pattern** with Azure Cosmos DB. We made use of some of the _hero features_ of Cosmos DB to simplify the implementation of the pattern.

You can find the source code for the sample application including the change feed processor on the [Microsoft Patterns & Practices repo](https://github.com/mspnp/transactional-outbox-pattern). There, you’ll also find a bicep deployment script to create all the necessary Azure resources for you automatically.

Keep in mind that this is not a “production ready” implementation, but it hopefully can act as an inspiration for your next adventure.

Happy hacking! 😊

> Implement the Transactional Outbox Pattern with @AzureCosmosDB and @Azure #ServiceBus #distributed #pattern #microservices
> 
> [Tweet](http://twitter.com/share?&text=Implement%20the%20Transactional%20Outbox%20Pattern%20with%20%40AzureCosmosDB%20and%20%40Azure%20%23ServiceBus%20%23distributed%20%23pattern%20%23microservices&url=https://partlycloudy.blog/2021/08/30/using-the-transactional-outbox-pattern-with-azure-cosmos-db-for-guaranteed-delivery-of-domain-events/&via=chrisdennig)

## References

- [1] – Jimmy Bogard – [A better domain events pattern](https://lostechies.com/jimmybogard/2014/05/13/a-better-domain-events-pattern/)
- [2] – Cosmos DB optimistic concurrency control – [https://docs.microsoft.com/en-us/azure/cosmos-db/sql/database-transactions-optimistic-concurrency#optimistic-concurrency-control](https://docs.microsoft.com/en-us/azure/cosmos-db/sql/database-transactions-optimistic-concurrency#optimistic-concurrency-control)
- [3] – Cosmos DB Time-To-Live feature – [https://docs.microsoft.com/en-us/azure/cosmos-db/sql/time-to-live](https://docs.microsoft.com/en-us/azure/cosmos-db/sql/time-to-live)
- [4] – Change Feed – Unsung Hero of Azure Cosmos DB – [https://devblogs.microsoft.com/cosmosdb/change-feed-unsung-hero-of-azure-cosmos-db/](https://devblogs.microsoft.com/cosmosdb/change-feed-unsung-hero-of-azure-cosmos-db/)
- [5] – Change Feed Processor library – [https://docs.microsoft.com/en-us/azure/cosmos-db/sql/change-feed-processor](https://docs.microsoft.com/en-us/azure/cosmos-db/sql/change-feed-processor)
- [6] – Azure Service Bus – Deduplication – [https://docs.microsoft.com/en-us/azure/service-bus-messaging/duplicate-detection](https://docs.microsoft.com/en-us/azure/service-bus-messaging/duplicate-detection)
- [7] – Azure Service Bus – FIFO – [https://docs.microsoft.com/en-us/azure/service-bus-messaging/message-sessions](https://docs.microsoft.com/en-us/azure/service-bus-messaging/message-sessions)
- Header Image credits: [https://unsplash.com/@pawel\_czerwinski](https://unsplash.com/@pawel_czerwinski)

## Additional Links

- Cosmos DB Transactional Batch – [https://docs.microsoft.com/en-us/azure/cosmos-db/transactional-batch](https://docs.microsoft.com/en-us/azure/cosmos-db/transactional-batch)
- Mediator implementation by Jimmy Bogard – [MediatR](https://github.com/jbogard/MediatR)
- Domain-Driven-Design – [https://docs.microsoft.com/en-us/dotnet/architecture/microservices/microservice-ddd-cqrs-patterns/](https://docs.microsoft.com/en-us/dotnet/architecture/microservices/microservice-ddd-cqrs-patterns/)

## Source Code

The sample source code for this article can be found on GitHub: [https://github.com/mspnp/transactional-outbox-pattern](https://github.com/mspnp/transactional-outbox-pattern).

* * *