Outbox with NHibernate Persistence

The outbox feature requires persistent storage in order to store the messages and enable deduplication.

Table

To keep track of duplicate messages, the NHibernate implementation of the outbox requires the creation of an OutboxRecord table.

The name of the table and the schema where the table is placed can be customized using the following API:

var persistence = endpointConfiguration.UsePersistence<NHibernatePersistence>();
persistence.CustomizeOutboxTableName(
    outboxTableName: "MyEndpointOutbox",
    outboxSchemaName: "MySchema");

Concurrency control

By default the outbox uses optimistic concurrency control. That means that when two copies of the same message arrive at the endpoint, both messages are picked up (if concurrency settings of the endpoint allow for it) and processing begins on both of them. When the message handlers are completed, both processing threads attempt to insert the outbox record as part of the transaction that includes the application state change.

At this point of of the transactions succeeds and the other fails due to unique index constraint violation. When the copy of the message that failed is picked up again, it is discarded as a duplicate.

The outcome is that the application state change is applied only once (the other attempt has been rolled back) but the message handlers have been executed twice. If the message handler contains logic that has non-transactional side effects (e.g. sending an e-mail), that logic may be executed multiple times.

Pessimistic concurrency control

The pessimistic concurrency control mode can be activated using the following API:

var outboxSettings = endpointConfiguration.EnableOutbox();
outboxSettings.UsePessimisticConcurrencyControl();

In the pessimistic mode the outbox record is inserted before the handlers are executed. As a result, when using a database that creates locks on insert, only one thread is allowed to execute the message handlers. The other thread, even though it picked up the second copy of a message, is blocked on a database lock. Once the first thread commits the transaction, the second thread is interrupted with an exception as it is not allowed to insert the outbox. As a result, the message handlers are executed only once.

The trade-off is that each message processing attempt requires additional round trip to the database.

Transactions

By default the outbox uses the ADO.NET transactions abstracted via ITransaction. This is appropriate for most situations.

Transaction Scope

In cases where the outbox transaction spans multiple databases, the TransactionScope support has to be enabled:

var outboxSettings = endpointConfiguration.EnableOutbox();
outboxSettings.UseTransactionScope();

In this mode the NHibernate persistence creates a TransactionScope that wraps the whole message processing attempt and within that scope it opens a session, that is used for:

• storing the outbox record
• persisting the application state change applied via SynchronizedStorageSession

In addition to that session managed by NServiceBus, users can open their own NHibernate sessions or plain database connections in the message handlers. If the underlying database technology supports distributed transactions managed by Microsoft Distributed Transaction Coordinator -- MS DTC (e.g. SQL Server, Oracle or PostgreSQL), the transaction gets escalated to a distributed transaction.

The TransactionScope mode is most useful in legacy scenarios e.g. when migrating from MSMQ transport to a messaging infrastructure that does not support MS DTC. In order to maintain consistency the outbox has to be used in place of distributed transport-database transactions. If the legacy database cannot be modified to add the outbox table, the only option is to place the outbox table in a separate database and use distributed transactions between the databases.

Customizing the transaction isolation level

The transaction isolation level for the outbox operation can be specified:

var outboxSettings = endpointConfiguration.EnableOutbox();
outboxSettings.UseTransactionScope();
outboxSettings.TransactionIsolationLevel(IsolationLevel.ReadCommitted);

Customizing outbox record persistence

By default the outbox records are persisted in the following way:

• The table has an auto-incremented integer primary key.
• The MessageId column has a unique index.
• There are indices on Dispatched and DispatchedAt columns.

The following API can be used to provide a different mapping of outbox data to the underlying storage:

var persistence = endpointConfiguration.UsePersistence<NHibernatePersistence>();
persistence.UseOutboxRecord<MyOutboxRecord, MyOutboxRecordMapping>();

public class MyOutboxRecord :
    IOutboxRecord
{
    public virtual string MessageId { get; set; }
    public virtual bool Dispatched { get; set; }
    public virtual DateTime? DispatchedAt { get; set; }
    public virtual string TransportOperations { get; set; }
}

public class MyOutboxRecordMapping :
    ClassMapping<MyOutboxRecord>
{
    public MyOutboxRecordMapping()
    {
        Table("MyOutboxTable");
        Id(
            idProperty: record => record.MessageId,
            idMapper: mapper => mapper.Generator(Generators.Assigned));
        Property(
            property: record => record.Dispatched,
            mapping: mapper =>
            {
                mapper.Column(c => c.NotNullable(true));
                mapper.Index("OutboxRecord_Dispatched_Idx");
            });
        Property(
            property: record => record.DispatchedAt,
            mapping: pm => pm.Index("OutboxRecord_DispatchedAt_Idx"));
        Property(
            property: record => record.TransportOperations,
            mapping: mapper => mapper.Type(NHibernateUtil.StringClob));
    }
}

If custom mapping is required, the following characteristics of the original mapping must be preserved:

• Values stored in the MessageId column must be unique and an attempt to insert a duplicate entry must cause an exception.
• Querying by Dispatched and DispatchedAt columns must be efficient because these columns are used by the cleanup process to remove outdated records.

Deduplication record lifespan

By default, the NHibernate implementation keeps deduplication records for seven days and checks for purgeable records every minute.

Specify different values in the config file using timestamp strings:

<appSettings>
  <add key="NServiceBus/Outbox/NHibernate/TimeToKeepDeduplicationData"
       value="7.00:00:00" />
  <add key="NServiceBus/Outbox/NHibernate/FrequencyToRunDeduplicationDataCleanup"
       value="00:01:00" />
</appSettings>

By specifying a value of -00:00:00.001 (i.e. 1 millisecond, the value of Timeout.InfiniteTimeSpan) for the NServiceBus/Outbox/NHibernate/FrequencyToRunDeduplicationDataCleanup app settings, the cleanup task is disabled. This is useful when an endpoint is scaled out and instances are competing to run the cleanup task.