Dynamo Outbox

Reference · Applies to Brighter V10

DynamoDb Outbox Usage

The DynamoDb Outbox allows integration between DynamoDb and Brighter's outbox support. The configuration is described in Command Processor Configuration Reference.

To support transactional messaging when using DynamoDb requires us to use DynamoDb's support for ACID transactions. You should understand best practices for using transactions with DynamoDb.

For this we will need the Outbox package for DynamoDb:

AWS SDK v3 (legacy support):

• Paramore.Brighter.Outbox.DynamoDB

AWS SDK v4 (recommended for new projects):

• Paramore.Brighter.Outbox.DynamoDB.V4

Paramore.Brighter.Outbox.DynamoDb (or .V4) will pull in another package:

• Paramore.Brighter.DynamoDb (or Paramore.Brighter.DynamoDb.V4)

The examples on this page use the v3 packages' namespaces. The v4 packages append .V4 to theirs: Paramore.Brighter.Outbox.DynamoDB.V4 and Paramore.Brighter.DynamoDb.V4, which also holds DynamoDbTableFactory.

See AWS SQS Migration for migration guidance between v3 and v4.

As described in Command Processor Configuration Reference, we configure Brighter to use an outbox by setting Outbox on the options passed to AddProducers().

As we want to use DynamoDb with the outbox, we also set ConnectionProvider and TransactionProvider — both to DynamoDbUnitOfWork, which is one type playing both parts — so that we can share your transaction scope when persisting messages to the outbox.

using System;
using Amazon.DynamoDBv2;
using Microsoft.Extensions.DependencyInjection;
using Paramore.Brighter;
using Paramore.Brighter.DynamoDb;
using Paramore.Brighter.Extensions.DependencyInjection;
using Paramore.Brighter.Outbox.DynamoDB;
using Paramore.Brighter.Outbox.Hosting;

public void ConfigureServices(IServiceCollection services)
{
    // ... dynamoDb is your IAmazonDynamoDB client, producerRegistry your transport
    services.AddSingleton<IAmazonDynamoDB>(dynamoDb);

    services.AddBrighter()
        .AddProducers(configure =>
        {
            configure.ProducerRegistry = producerRegistry;
            configure.Outbox = new DynamoDbOutbox(
                dynamoDb, new DynamoDbConfiguration(), TimeProvider.System);
            configure.ConnectionProvider = typeof(DynamoDbUnitOfWork);
            configure.TransactionProvider = typeof(DynamoDbUnitOfWork);
        })
        .UseOutboxSweeper()
        .AutoFromAssemblies();
}

The handler below is AddGreetingHandlerAsync from the Brighter sample at Brighter/samples/WebAPI/WebAPI_Dynamo/, which also declares the AddGreeting request, the Person entity and the GreetingMade event it uses.

In our handler we take a dependency on Brighter's IAmADynamoDbTransactionProvider interface, which DynamoDbUnitOfWork implements. We explicitly start a transaction within the handler on the Database within that provider.

We call DepositPostAsync within that transaction to write the message to the Outbox. Once the transaction has closed we can call ClearOutboxAsync to immediately clear, or we can rely on the Outbox Sweeper, if we have configured one to clear for us. (There are equivalent synchronous versions of these APIs).

Running more than one instance? Configure a distributed lock so only one Sweeper (and Archiver) runs at a time — see DynamoDB Distributed Lock.

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Amazon.DynamoDBv2.DataModel;
using Amazon.DynamoDBv2.Model;
using Microsoft.Extensions.Logging;
using Paramore.Brighter;
using Paramore.Brighter.DynamoDb;

public class AddGreetingHandlerAsync : RequestHandlerAsync<AddGreeting>
{
    private readonly IAmADynamoDbTransactionProvider _transactionProvider;
    private readonly IAmACommandProcessor _postBox;
    private readonly ILogger<AddGreetingHandlerAsync> _logger;

    public AddGreetingHandlerAsync(IAmADynamoDbTransactionProvider transactionProvider,
        IAmACommandProcessor postBox,
        ILogger<AddGreetingHandlerAsync> logger)
    {
        _transactionProvider = transactionProvider;
        _postBox = postBox;
        _logger = logger;
    }

    public override async Task<AddGreeting> HandleAsync(AddGreeting addGreeting, CancellationToken cancellationToken = default)
    {
        var posts = new List<Id>();

        //We use the transaction provider to grab connection and transaction, because Outbox needs
        //to share them 'behind the scenes'
        var context = new DynamoDBContext(_transactionProvider.DynamoDb);
        var transaction = await _transactionProvider.GetTransactionAsync(cancellationToken);
        try
        {
            var person = await context.LoadAsync<Person>(addGreeting.Name, cancellationToken);

            person.Greetings.Add(addGreeting.Greeting);

            var document = context.ToDocument(person);
            var attributeValues = document.ToAttributeMap();

            //write the added child entity to the Db - just replace the whole entity as we grabbed the original
            //in production code, an update expression would be faster
            transaction.TransactItems.Add(new TransactWriteItem{Put = new Put{TableName = "People", Item = attributeValues}});

            //Now write the message we want to send to the Db in the same transaction.
            posts.Add(await _postBox.DepositPostAsync(
                new GreetingMade(addGreeting.Greeting),
                _transactionProvider,
                cancellationToken: cancellationToken));

            //commit both new greeting and outgoing message
            await _transactionProvider.CommitAsync(cancellationToken);
        }
        catch (Exception e)
        {
            _logger.LogError(e, "Exception thrown handling Add Greeting request");
            //it went wrong, rollback the entity change and the downstream message
            _transactionProvider.Rollback();
            return await base.HandleAsync(addGreeting, cancellationToken);
        }

        //Send this message via a transport. We need the ids to send just the messages here, not all outstanding ones.
        //Alternatively, you can let the Sweeper do this, but at the cost of increased latency
        await _postBox.ClearOutboxAsync(posts, cancellationToken: cancellationToken);

        return await base.HandleAsync(addGreeting, cancellationToken);
    }
}

DynamoDb Outbox Options

DynamoDbConfiguration names the table and its four global secondary indexes, and carries the two settings that govern expiry and scan concurrency.

Option	Type	Default	Description
TableName	string	"brighter_outbox"	Names the DynamoDB table that holds the Outbox.
DeliveredIndexName	string	"Delivered"	Names the global secondary index of dispatched messages, indexed by topic.
DeliveredAllTopicsIndexName	string	"DeliveredAllTopics"	Names the global secondary index of dispatched messages covering every topic.
OutstandingIndexName	string	"Outstanding"	Names the global secondary index of undispatched messages, indexed by topic.
OutstandingAllTopicsIndexName	string	"OutstandingAllTopics"	Names the global secondary index of undispatched messages covering every topic.
TimeToLive	TimeSpan?	null	Sets how long a message survives in the table before DynamoDB expires it; messages do not expire when it is null.
ScanConcurrency	int	3	Sets how many segments a parallel scan for outstanding messages uses.

timeout and numberOfShards are constructor parameters, not properties you can set. They surface as the get-only Timeout and NumberOfShards. numberOfShards defaults to 3 and shards the outstanding index so an active topic does not build a hot partition; the Outbox throws ArgumentOutOfRangeException above 20. timeout defaults to 500 and nothing reads it; neither does the DynamoDB Outbox read the outBoxTimeout argument the IAmAnOutbox methods take, which it accepts and passes between overloads without ever acting on. Configure timeouts on the AmazonDynamoDBClient you hand to the Outbox instead. tableName and scanConcurrency are constructor parameters as well, but each has a settable property and is in the table above.

Replay Support: The Causation Index

Not in a released package yet. This section describes Replay On Seen, which ships after Brighter 10.7.0 — the current release. DynamoDbConfiguration.CausationIndexName and the CausationId attribute below are on Brighter's development branch and are in no version you can install today, which is why CausationIndexName is absent from the options table above.

Replay On Seen resends every Outbox message produced under a given Causation Id. On DynamoDB that means querying on a non-key attribute, which needs a Global Secondary Index — by default one named Causation, with CausationId as its hash key.

This applies to both Paramore.Brighter.Outbox.DynamoDB and .V4. The DynamoDB Inbox needs nothing: it looks a Causation Id up by the table's own primary key.

A new table gets the index for free

MessageItem.CausationId is decorated with [DynamoDBGlobalSecondaryIndexHashKey(indexName: "Causation")], and DynamoDbTableFactory reflects over those attributes when it builds the request. So a table you create through the factory already has the index — you only add a throughput entry for it:

using System.Collections.Generic;
using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.Model;
using Paramore.Brighter.DynamoDb;
using Paramore.Brighter.Outbox.DynamoDB;

var createTableRequest = new DynamoDbTableFactory().GenerateCreateTableRequest<MessageItem>(
    new DynamoDbCreateProvisionedThroughput(
        new ProvisionedThroughput { ReadCapacityUnits = 10, WriteCapacityUnits = 10 },
        new Dictionary<string, ProvisionedThroughput?>
        {
            ["Outstanding"] = new() { ReadCapacityUnits = 10, WriteCapacityUnits = 10 },
            ["OutstandingAllTopics"] = new() { ReadCapacityUnits = 10, WriteCapacityUnits = 10 },
            ["Delivered"] = new() { ReadCapacityUnits = 10, WriteCapacityUnits = 10 },
            ["DeliveredAllTopics"] = new() { ReadCapacityUnits = 10, WriteCapacityUnits = 10 },
            ["Causation"] = new() { ReadCapacityUnits = 10, WriteCapacityUnits = 10 },  // replay
        }));

var builder = new DynamoDbTableBuilder(client);
await builder.Build(createTableRequest);
await builder.EnsureTablesReady([createTableRequest.TableName], TableStatus.ACTIVE);

Adding the index to an existing table

A table provisioned before replay shipped does not have the index, and there is no migration runner for DynamoDB. Add it yourself with UpdateTable:

using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.Model;

var request = new UpdateTableRequest
{
    TableName = "brighter_outbox",

    // The index's key attribute has to be declared, even though the table already stores it
    AttributeDefinitions =
    [
        new AttributeDefinition
        {
            AttributeName = "CausationId",
            AttributeType = ScalarAttributeType.S
        }
    ],

    GlobalSecondaryIndexUpdates =
    [
        new GlobalSecondaryIndexUpdate
        {
            Create = new CreateGlobalSecondaryIndexAction
            {
                IndexName = "Causation",
                KeySchema =
                [
                    new KeySchemaElement { AttributeName = "CausationId", KeyType = KeyType.HASH }
                ],

                // Replay only reads MessageId, which is the base table's hash key and is
                // therefore always present in a KEYS_ONLY index
                Projection = new Projection { ProjectionType = ProjectionType.KEYS_ONLY },

                // Omit this on a PAY_PER_REQUEST table — it applies to PROVISIONED billing only
                ProvisionedThroughput = new ProvisionedThroughput
                {
                    ReadCapacityUnits = 10,
                    WriteCapacityUnits = 10
                }
            }
        }
    ]
};

await client.UpdateTableAsync(request);

Two operational notes:

• The index backfills asynchronously. DynamoDB populates a new GSI in the background, and the index reports CREATING until it finishes. Until then a replay may find only part of a causation's messages, so wait for the index to become ACTIVE before you rely on it.

• Restart your hosts afterwards. The Outbox probes for the index once, with DescribeTable, and caches the answer for the life of the store instance. An Outbox built before the index existed keeps reporting that replay is unsupported until the process restarts.

The index name

DynamoDbConfiguration.CausationIndexName defaults to "Causation". Leave it there.

It has a public setter, unlike the name suggests it should be treated — but it does not behave like the Outstanding and Delivered index names, which you can rename freely. The Causation index name is also declared on MessageItem as an attribute argument, and attribute arguments must be compile-time constants, so the annotation cannot read your configured value. DynamoDbTableFactory therefore always generates an index called Causation, whatever you set here.

Point CausationIndexName at another name and the probe and the replay query both target a GSI the table model never declares. Nothing errors: SupportsCausationTracking() simply reports false, and replay silently finds no messages. If you have a genuine reason to rename it, you must create the index under that name yourself.

If you skip the index

Nothing breaks, and replay never happens:

• Deposits are unaffected. The CausationId attribute is still written; with no index to populate it is simply an ordinary attribute.

• Startup warns. SupportsCausationTracking() reports false, and pipeline validation raises a warning — not an error — for any handler configured with OnceOnlyAction.Replay.

• Duplicates are skipped quietly. ReplayCausation returns false rather than throwing, so a duplicate does not fail the consumer pipeline with a DynamoDB ValidationException. Nothing is resent.

See When Replay Does Not Fire for how to tell this apart from the other reasons a replay produces nothing.