|
| 1 | +# Transactional Outbox Plugin for MongoDB <!-- omit in toc --> |
| 2 | + |
| 3 | +Please read the [Introduction](intro.md) and the [Transactional Outbox](plugin_outbox.md) overview before reading this page. |
| 4 | + |
| 5 | +- [Introduction](#introduction) |
| 6 | +- [Configuration](#configuration) |
| 7 | +- [Options](#options) |
| 8 | + - [UseOutbox for Producers](#useoutbox-for-producers) |
| 9 | + - [UseMongoDbTransaction for Consumers](#usemongodbtransaction-for-consumers) |
| 10 | +- [How it works](#how-it-works) |
| 11 | +- [Collections](#collections) |
| 12 | +- [Migration versioning](#migration-versioning) |
| 13 | +- [Indices](#indices) |
| 14 | +- [Clean up](#clean-up) |
| 15 | +- [Important note](#important-note) |
| 16 | + |
| 17 | +## Introduction |
| 18 | + |
| 19 | +[`SlimMessageBus.Host.Outbox.MongoDb`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.MongoDb) adds [Transactional Outbox](https://microservices.io/patterns/data/transactional-outbox.html) pattern support backed by **MongoDB**. |
| 20 | + |
| 21 | +It uses the [MongoDB.Driver](https://www.nuget.org/packages/MongoDB.Driver) (3.x) and targets **.NET 8 and .NET 10**. |
| 22 | + |
| 23 | +> Requires `IMongoClient` to be registered in the DI container. |
| 24 | +
|
| 25 | +## Configuration |
| 26 | + |
| 27 | +> Required: [`SlimMessageBus.Host.Outbox.MongoDb`](https://www.nuget.org/packages/SlimMessageBus.Host.Outbox.MongoDb) |
| 28 | +
|
| 29 | +```bash |
| 30 | +dotnet add package SlimMessageBus.Host.Outbox.MongoDb |
| 31 | +``` |
| 32 | + |
| 33 | +Call `.AddOutboxUsingMongoDb()` on the `MessageBusBuilder` to enable the plugin: |
| 34 | + |
| 35 | +```csharp |
| 36 | +using SlimMessageBus.Host.Outbox.MongoDb; |
| 37 | + |
| 38 | +builder.Services.AddSlimMessageBus(mbb => |
| 39 | +{ |
| 40 | + mbb |
| 41 | + .AddChildBus("Memory", mbb => |
| 42 | + { |
| 43 | + mbb.WithProviderMemory() |
| 44 | + .AutoDeclareFrom(Assembly.GetExecutingAssembly(), consumerTypeFilter: t => t.Name.EndsWith("CommandHandler")) |
| 45 | + // Wrap each command handler in a MongoDB multi-document transaction |
| 46 | + .UseMongoDbTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); |
| 47 | + }) |
| 48 | + .AddChildBus("AzureSB", mbb => |
| 49 | + { |
| 50 | + mbb.WithProviderServiceBus(cfg => { /* ... */ }) |
| 51 | + .Produce<CustomerCreatedEvent>(x => x.DefaultTopic("samples.outbox/customer-events")) |
| 52 | + // All outgoing messages from this bus will go out via the outbox |
| 53 | + .UseOutbox(); |
| 54 | + }) |
| 55 | + .AddServicesFromAssembly(Assembly.GetExecutingAssembly()) |
| 56 | + .AddJsonSerializer() |
| 57 | + // Configure MongoDB outbox |
| 58 | + .AddOutboxUsingMongoDb(opts => |
| 59 | + { |
| 60 | + opts.PollBatchSize = 500; |
| 61 | + opts.PollIdleSleep = TimeSpan.FromSeconds(10); |
| 62 | + opts.MessageCleanup.Interval = TimeSpan.FromSeconds(60); |
| 63 | + opts.MessageCleanup.Age = TimeSpan.FromMinutes(60); |
| 64 | + // Override MongoDB collection names (optional) |
| 65 | + // opts.MongoDbSettings.DatabaseName = "myapp"; |
| 66 | + // opts.MongoDbSettings.CollectionName = "smb_outbox"; |
| 67 | + // opts.MongoDbSettings.LockCollectionName = "smb_outbox_lock"; |
| 68 | + }); |
| 69 | +}); |
| 70 | + |
| 71 | +// SMB requires IMongoClient to be registered in the container |
| 72 | +builder.Services.AddSingleton<IMongoClient>(new MongoClient(connectionString)); |
| 73 | +``` |
| 74 | + |
| 75 | +## Options |
| 76 | + |
| 77 | +### UseOutbox for Producers |
| 78 | + |
| 79 | +`.UseOutbox()` marks a producer (or an entire child bus) to route outgoing messages through the outbox instead of publishing them directly to the transport. |
| 80 | + |
| 81 | +```csharp |
| 82 | +mbb.Produce<OrderCreatedEvent>(x => |
| 83 | +{ |
| 84 | + x.DefaultTopic("order-events"); |
| 85 | + x.UseOutbox(); // this producer uses the outbox |
| 86 | +}); |
| 87 | + |
| 88 | +// or for all producers on a bus: |
| 89 | +mbb.UseOutbox(); |
| 90 | +``` |
| 91 | + |
| 92 | +### UseMongoDbTransaction for Consumers |
| 93 | + |
| 94 | +`.UseMongoDbTransaction()` wraps each consumer (or handler) in a MongoDB multi-document transaction. The transaction is committed after a successful `OnHandle` call and rolled back on any exception. |
| 95 | + |
| 96 | +> **Note:** MongoDB multi-document transactions require a **replica set** (or sharded cluster). Standalone `mongod` instances do not support transactions. |
| 97 | +
|
| 98 | +```csharp |
| 99 | +using SlimMessageBus.Host.Outbox.MongoDb; |
| 100 | + |
| 101 | +// On a single consumer: |
| 102 | +mbb.Consume<CreateCustomerCommand>(x => |
| 103 | + x.WithConsumer<CreateCustomerCommandHandler>() |
| 104 | + .UseMongoDbTransaction()); |
| 105 | + |
| 106 | +// Or across all consumers on a bus: |
| 107 | +mbb.UseMongoDbTransaction(messageTypeFilter: t => t.Name.EndsWith("Command")); |
| 108 | +``` |
| 109 | + |
| 110 | +#### Enlisting your own MongoDB writes in the transaction |
| 111 | + |
| 112 | +The **outbox insert** always participates in the active transaction automatically. However, unlike SQL (where a `SqlConnection` carries the transaction implicitly), MongoDB requires the `IClientSessionHandle` to be **passed explicitly** to every collection operation. |
| 113 | + |
| 114 | +To make your own document writes atomic with the outbox insert, inject `IClientSessionHandle?` directly into the consumer constructor: |
| 115 | + |
| 116 | +```csharp |
| 117 | +// No dependency on SlimMessageBus.Host.Outbox.MongoDb — only MongoDB.Driver types needed. |
| 118 | +public class CreateCustomerCommandHandler( |
| 119 | + IMongoCollection<Customer> customers, |
| 120 | + IClientSessionHandle? session, // null when no transaction is active |
| 121 | + IMessageBus bus) : IRequestHandler<CreateCustomerCommand, Guid> |
| 122 | +{ |
| 123 | + public async Task<Guid> OnHandle(CreateCustomerCommand request, CancellationToken ct) |
| 124 | + { |
| 125 | + var customer = new Customer(request.Name); |
| 126 | + |
| 127 | + // Both writes share the same session — committed or rolled back together. |
| 128 | + if (session != null) |
| 129 | + await customers.InsertOneAsync(session, customer, cancellationToken: ct); |
| 130 | + else |
| 131 | + await customers.InsertOneAsync(customer, cancellationToken: ct); |
| 132 | + |
| 133 | + // This publish goes via the outbox and is in the same transaction. |
| 134 | + await bus.Publish(new CustomerCreatedEvent(customer.Id)); |
| 135 | + return customer.Id; |
| 136 | + } |
| 137 | +} |
| 138 | +``` |
| 139 | + |
| 140 | +> **Why does constructor injection work here?** |
| 141 | +> SMB resolves the consumer from DI *after* all interceptors have executed. `MongoDbTransactionConsumerInterceptor` starts the session before the consumer is constructed, so the DI factory for `IClientSessionHandle` already finds a live session in `MongoDbSessionHolder` by the time the consumer's constructor runs. See [Consumer instance resolution order](intro.md#consumer-instance-resolution-order) for the full execution diagram. |
| 142 | +
|
| 143 | +`session` is `null` when no transaction is active (e.g. `UseMongoDbTransaction()` is not configured, or running against a standalone `mongod`). The `null` check makes the consumer work in both cases. |
| 144 | + |
| 145 | +## How it works |
| 146 | + |
| 147 | +- On bus start, `MongoDbOutboxMigrationService` creates the outbox collection and lock collection (if they do not exist) together with the supporting indices. |
| 148 | +- When a message is published via a producer marked with `.UseOutbox()`, the message is inserted into the outbox MongoDB collection. |
| 149 | + - If the call happens inside a consumer that has `.UseMongoDbTransaction()` enabled, the insert participates in the active MongoDB session, ensuring atomicity with any other writes performed during that consumer invocation. |
| 150 | +- A background poller periodically locks a batch of undelivered messages (up to `PollBatchSize`) and forwards them to the actual transport. Locking is done in two steps: |
| 151 | + 1. Find candidate document IDs (ordered by `Timestamp`, limited to `PollBatchSize`). |
| 152 | + 2. Atomically claim them with an `UpdateMany` that re-applies the eligibility filter to handle concurrent instances. |
| 153 | +- When `MaintainSequence = true`, an additional global lock document (in the lock collection) ensures only one application instance processes the outbox at a time, preserving message order at the cost of throughput. |
| 154 | +- After successful delivery each document is marked `DeliveryComplete = true`. On repeated failures the `DeliveryAttempt` counter is incremented; once it reaches `MaxDeliveryAttempts` the document is marked `DeliveryAborted = true` and skipped. |
| 155 | + |
| 156 | +## Collections |
| 157 | + |
| 158 | +By default three MongoDB collections are used: |
| 159 | + |
| 160 | +| Collection | Setting | Default | |
| 161 | +| ----------------------- | ------------------------------------------ | ------------------------ | |
| 162 | +| Outbox messages | `MongoDbSettings.CollectionName` | `smb_outbox` | |
| 163 | +| Global lock (table-lock mode) | `MongoDbSettings.LockCollectionName` | `smb_outbox_lock` | |
| 164 | +| Applied migrations | `MongoDbSettings.MigrationsCollectionName` | `smb_outbox_migrations` | |
| 165 | + |
| 166 | +The database is set via `MongoDbSettings.DatabaseName` (default: `slimmessagebus`). |
| 167 | + |
| 168 | +## Migration versioning |
| 169 | + |
| 170 | +Schema changes are tracked in the `smb_outbox_migrations` collection. Each migration step has a unique timestamp-based ID (e.g. `"20240101000000_SMB_Init"`). On startup `MongoDbOutboxMigrationService` checks whether each migration ID is present in the collection: |
| 171 | + |
| 172 | +- **Not present** → the action (index creation/modification) runs and the ID is recorded on success. |
| 173 | +- **Present** → skipped. |
| 174 | + |
| 175 | +This gives **at-least-once** (not exactly-once) execution semantics: |
| 176 | + |
| 177 | +- A crash before the record is written → **retried on the next startup** (safe, all actions are idempotent). |
| 178 | +- Two instances racing simultaneously → both may run the action, one wins the insert race, the other handles the `DuplicateKey` exception (safe for idempotent actions). |
| 179 | + |
| 180 | +> **Note:** MongoDB does not allow DDL operations such as `createIndex` inside multi-document transactions. Migrations are therefore intentionally **not transactional** — safety comes from idempotency, not atomicity. Only add migration steps that are safe to run more than once (i.e. index creation using `IF NOT EXISTS` semantics). Destructive one-shot operations must be applied externally with `EnableMigration = false`. |
| 181 | +
|
| 182 | +To add a future migration, append a new `TryApplyMigration` call in the service with a new unique ID. Old migration IDs must never be reused. |
| 183 | + |
| 184 | +### Disabling migrations |
| 185 | + |
| 186 | +Set `MongoDbSettings.EnableMigration = false` to skip the entire migration step at startup. Use this when you manage schema changes externally (e.g. via a deployment pipeline) and want SMB to leave the database schema untouched. |
| 187 | + |
| 188 | +```csharp |
| 189 | +.AddOutboxUsingMongoDb(opts => |
| 190 | +{ |
| 191 | + opts.MongoDbSettings.EnableMigration = false; |
| 192 | +}); |
| 193 | +``` |
| 194 | + |
| 195 | +## Indices |
| 196 | + |
| 197 | +`MongoDbOutboxMigrationService` ensures the following indices exist on startup: |
| 198 | + |
| 199 | +**Outbox collection (`smb_outbox`)** |
| 200 | + |
| 201 | +| Index fields | Purpose | |
| 202 | +| --------------------------------------------------- | -------------------------------- | |
| 203 | +| `delivery_complete`, `delivery_aborted`, `timestamp` | Main polling query | |
| 204 | +| `lock_instance_id`, `lock_expires_on` | Lock-ownership queries | |
| 205 | +| `timestamp` | Cleanup (delete-sent) ordering | |
| 206 | + |
| 207 | +**Lock collection (`smb_outbox_lock`)** |
| 208 | + |
| 209 | +| Index fields | Purpose | |
| 210 | +| ---------------- | --------------------- | |
| 211 | +| `lock_expires_on` | Expired-lock detection | |
| 212 | + |
| 213 | +## Clean up |
| 214 | + |
| 215 | +Sent messages older than `MessageCleanup.Age` are removed in batches of `MessageCleanup.BatchSize` on startup and then every `MessageCleanup.Interval`. |
| 216 | + |
| 217 | +| Property | Description | Default | |
| 218 | +| --------- | -------------------------------------------------- | ------- | |
| 219 | +| Enabled | `true` if sent messages are to be removed | true | |
| 220 | +| Interval | Time between clean-up executions | 1 hour | |
| 221 | +| Age | Minimum age of a sent message to delete | 1 hour | |
| 222 | +| BatchSize | Number of messages to be removed in each iteration | 10 000 | |
| 223 | + |
| 224 | +## Important note |
| 225 | + |
| 226 | +Because the outbox can be processed by any application instance, all active instances must share the same message registrations and compatible serialization schema. |
| 227 | + |
| 228 | +A message that fails to be delivered will have its `DeliveryAborted` flag set to `true` in the outbox collection once `MaxDeliveryAttempts` is exceeded. It is safe to reset this flag to `false` manually (e.g. via `mongosh`) once the underlying issue has been resolved. |
0 commit comments