MongoDB transactions

The MongoDB path carries a session opened through the same DI DatabaseService into every database operation. A transaction-capable replica set or sharded deployment is required; standalone MongoDB is not sufficient.

Feature and queue binding#

[dependencies]
lilyrs = { version = "0.1.0", features = ["consumer-transactional-inbox-mongodb", "config"] }
serde = { version = "1", features = ["derive"] }
uuid = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tokio-util = "0.7"

This manifest selects single-connection mode. Keep the connection, queue and retention settings from the first application; append the following table after the retention table of the orders.created queue. Omit database_cell in single mode. Use the separate features and examples for factory mode.

[rabbitmq.topology.queues.transactional_inbox]
backend = "mongodb"

Configure the MongoDB connection under Lily’s [database] section; the section is not named [mongodb]. The business collection and migrator must use the same DatabaseService.

[database]
mode = "single"
database_type = "mongodb"
connection_string = "${secret:mongodb.primary_url}"
database_name = "orders"

This top-level database section belongs outside the queue table. The example enables config; supply a secret resolver for ${secret:mongodb.primary_url}. The resolved URI must target a transaction-capable replica set or sharded deployment. The MongoDB integration describes additional connection and TLS options.

Handler and migration function#

The following file is a compilable application module without a main function. Include it in your first application and remove the previous example handler for the same queue. Call migrate from a separate deployment command with an initialized container, rather than from Consumer startup. That command owns creating and closing its container.

use std::sync::Arc;
use lilyrs::consumer::{ApplicationContainer, Injectable, ServiceTrait};
use lilyrs::mongodb::{Collection, DatabaseService, MongoCollection};
use lilyrs::queue::{Json, MongoInboxOutboxMigrator, MongoTransaction,
    PublishContentKind, QueueHandlerError, TransactionalOutboxMessage, queue, queue_service};
use serde::{Deserialize, Serialize};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

#[derive(Clone, Serialize, Deserialize)]
struct Order { order_id: i64 }
#[derive(Default, Injectable, MongoCollection)]
#[service(lifetime = "Singleton")]
#[collection("orders")]
#[collection_type(Order)]
struct Orders {
    #[inject]
    db: Arc<DatabaseService>,
    collection: Option<Collection<Order>>,
}
#[derive(Deserialize)]
struct OrderCreated { order_id: i64, outgoing_event_id: String }
#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct OrderWorker {
    #[inject]
    orders: Arc<Orders>,
}
impl ServiceTrait for OrderWorker {}

#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json", delivery_guarantee = "transactional_inbox")]
    async fn created(&self, transaction: MongoTransaction, Json(event): Json<OrderCreated>)
        -> Result<(), QueueHandlerError>
    {
        let outgoing = Uuid::parse_str(&event.outgoing_event_id)
            .map_err(|_| QueueHandlerError::permanent("ORDER_EVENT_ID_INVALID"))?;
        let operation = transaction.operation_context()?;
        self.orders.insert_one(Order { order_id: event.order_id }, &operation).await
            .map_err(|_| QueueHandlerError::retryable("ORDER_WRITE_FAILED"))?;
        transaction.enqueue(TransactionalOutboxMessage::try_new(
            outgoing, "domain.events", "orders.persisted", 1,
            PublishContentKind::Json, br#"{"status":"persisted"}"#.as_slice(),
        )?).await?;
        Ok(())
    }
}

// Apply application collection migrations before the Lily component migration.
pub async fn migrate(container: &ApplicationContainer) -> Result<(), Box<dyn std::error::Error>> {
    let database = container.resolve::<DatabaseService>(None).await?;
    let operation = database.operation_context(CancellationToken::new())?;
    MongoInboxOutboxMigrator::new(database.as_ref())?.apply(&operation).await?;
    Ok(())
}

Download the complete mongodb.rs example.

Business data and outgoing events#

transaction.operation_context() carries the session, deadline and cancellation into the collection operation. Creating a fresh context through ordinary DatabaseService::operation_context would put the handler write outside this transaction. Do not clone the context into detached work; its authority ends when the body finishes.

The example input JSON contains order_id and outgoing_event_id. The latter must be a valid non-nil UUID; the example uses it as the application-assigned identity of the outgoing event. It is separate from the incoming delivery’s canonical event ID. Preserve the identity when republishing the same logical outgoing event, and do not reuse it for different events.

{"order_id": 42, "outgoing_event_id": "e7b04866-8f2e-46e4-a92c-d35dd2016879"}

The outbox destination is the domain.events exchange with routing key orders.persisted. Provision that destination binding separately; publication needs a routable destination. Adding a destination queue to this Consumer’s topology without a matching local handler fails its startup contract checks.

For teaching purposes, the example maps a collection-write error to the retryable ORDER_WRITE_FAILED error. This does not mean every database error is transient. In a real application, choose an error policy that distinguishes validation, business uniqueness constraints and permanent data errors from transient connection problems.

MongoDB retries and leases#

TransientTransactionError can replay the entire business pipeline, including middleware before/after, guards, extractors and the handler, in a new transaction/session and delivery scope. max_transaction_attempts includes the first execution. If each attempt calls an external system, database rollback cannot undo that call.

For UnknownTransactionCommitResult, Lily retries the same commit; that label alone does not rerun the body. Commit reconciliation is bounded by both the MongoDB policy and the existing delivery deadline. If the result remains ambiguous, the lease is left to expire and the original delivery is directly requeued without consuming the application retry/DLQ budget.

Lease contention on the same inbox key waits within the existing absolute delivery budget and can requeue the original delivery. A lease-release error after commit is advisory: it does not turn an Applied or AlreadyCompleted result into a business failure. See the configuration reference for policy values.

Verify before starting#

Ordinary startup does not run migrations. Lily checks the previously installed schema contract through read-only preflight and prepares the database binding before queue admission. Missing migrations or schema drift prevent listener admission. This does not mean DI initializers have made no network connections. Migrations and retention describes the deployment sequence.