PostgreSQL transactions

The PostgreSQL path uses the PgDatabaseService pool in the Consumer application container. The inbox, business-table writes and outbox records share one PostgreSQL transaction; Lily does not open a second transaction pool.

Feature and queue binding#

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

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 = "postgresql"

Configure the PostgreSQL connection under Lily’s [postgresql] section. Your ordinary deployment migration creates the application’s orders table. The example writes to an id BIGINT field; choose its primary-key and uniqueness policy according to your business model.

[postgresql]
mode = "single"
connection_string = "${secret:postgresql.primary_url}"

[postgresql.tls]
mode = "verify-full"

This top-level database section belongs outside the queue table. The example enables config; supply a secret resolver for ${secret:postgresql.primary_url}. verify-full requires appropriate CA and hostname verification; follow the PostgreSQL integration for private CAs and connection options. The resolved connection string must contain credentials and the target database. Disabling TLS for local testing belongs in a separate development configuration; production validation requires verify-full.

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 lilyrs::consumer::{ApplicationContainer, Injectable, ServiceTrait};
use lilyrs::queue::{Json, PostgresInboxOutboxMigrator, PostgresTransaction,
    PublishContentKind, QueueHandlerError, TransactionalOutboxMessage, queue, queue_service};
use lilyrs::postgresql::{diesel, diesel_async::RunQueryDsl};
use serde::Deserialize;
use uuid::Uuid;

#[derive(Deserialize)]
struct OrderCreated { order_id: i64, outgoing_event_id: String }
#[derive(Default, Injectable)]
#[service(lifetime = "Scoped")]
struct OrderWorker;
impl ServiceTrait for OrderWorker {}

#[queue_service]
impl OrderWorker {
    #[queue("orders.created", version = 1, content = "json", delivery_guarantee = "transactional_inbox")]
    async fn created(&self, transaction: PostgresTransaction, Json(event): Json<OrderCreated>)
        -> Result<(), QueueHandlerError>
    {
        let outgoing = Uuid::parse_str(&event.outgoing_event_id)
            .map_err(|_| QueueHandlerError::permanent("ORDER_EVENT_ID_INVALID"))?;
        transaction.with_connection(move |connection| Box::pin(async move {
            // Create this application table in your deployment migration first.
            diesel::sql_query("INSERT INTO orders (id) VALUES ($1)")
                .bind::<diesel::sql_types::BigInt, _>(event.order_id)
                .execute(connection).await?;
            Ok(())
        })).await?;
        transaction.enqueue(TransactionalOutboxMessage::try_new(
            outgoing, "domain.events", "orders.persisted", 1,
            PublishContentKind::Json, br#"{"status":"persisted"}"#.as_slice(),
        )?).await?;
        Ok(())
    }
}

// Invoke from a separate deployment migration command, before Consumer startup.
pub async fn migrate(container: &ApplicationContainer) -> Result<(), Box<dyn std::error::Error>> {
    let database = container.resolve::<lilyrs::postgresql::PgDatabaseService>(None).await?;
    PostgresInboxOutboxMigrator::new(database).migrate().await?;
    Ok(())
}

Download the complete postgresql.rs example.

Business data and outgoing events#

A query inside the transaction.with_connection closure uses the same connection and transaction as the inbox. A separate pool query outside that closure is not covered. Handlers do not issue commit or rollback calls; Lily owns the transaction.

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.

Concurrent duplicates and retries#

If two deliveries concurrently handle the same handler/event key, PostgreSQL’s non-blocking advisory lock check turns InProgress into a retryable error. This follows the ordinary queue retry/DLQ policy. On a queue that expects concurrent duplicates, consider that retry_attempts = 0 leads directly to dead-letter handling.

After a successful commit, a repeated event skips business code through the completed inbox record. The PostgreSQL path does not use MongoDB’s transaction-label-driven pipeline replay mechanism. An application retry is a separate broker delivery and creates its own scope.

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.