PostgreSQL transaction’ları

PostgreSQL yolu, Consumer’ın DI container’ındaki PgDatabaseService havuzunu kullanır. Inbox, iş tablosu yazısı ve outbox kaydı aynı PostgreSQL transaction’ında tutulur; ikinci bir transaction havuzu açılmaz.

Feature ve 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"] }

Bu manifest tek bağlantı modu içindir. İlk uygulamadaki connection, queue ve retention ayarlarını koruyun; aşağıdaki tabloyu orders.created queue’sunun retention tablosundan sonra ekleyin. Tek bağlantı modunda database_cell yazılmaz. Factory modu için ayrı feature ve örnekleri kullanın.

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

PostgreSQL bağlantısını Lily’nin [postgresql] bölümünde yapılandırın. orders uygulama tablosunu normal deployment migration’ınız oluşturur. Örnek id BIGINT alanına yazar; tablonun primary key/unique politikasını iş modelinize göre seçin.

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

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

Bu top-level DB bölümü queue tablosunun dışındadır. config feature’ı örneğe eklenmiştir; ${secret:postgresql.primary_url} için secret resolver sağlayın. verify-full uygun CA ve hostname doğrulaması gerektirir; özel CA’nız ve bağlantı seçenekleri için PostgreSQL entegrasyonunu izleyin. Mevcut connection string’inizde credentials ve hedef DB bulunmalıdır. Local test için TLS’i kapatma kararı ayrı bir development config’ine aittir; production doğrulaması verify-full ister.

Handler ve migration fonksiyonu#

Aşağıdaki dosya derlenebilir bir uygulama modülüdür; main içermez. İlk uygulamanıza bu modülü ekleyin ve aynı queue için eski örnek handler’ı kaldırın. migrate fonksiyonunu Consumer startup’ı sırasında değil, hazır container kullanan ayrı deployment komutunuzdan çağırın. Container’ın oluşturulması ve kapatılması o komutun sorumluluğudur.

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(())
}

Tam postgresql.rs örneğini indirin.

İş verisi ve outgoing event#

transaction.with_connection closure’ı içindeki sorgu, inbox’ı yöneten aynı connection ve transaction üzerinde çalışır. Closure dışındaki ayrı bir pool query’si atomik kapsama girmez. Commit/rollback çağrılarını handler yazmaz; transaction owner’ı Lily’dir.

Gelen örnek JSON, order_id ve outgoing_event_id alanlarını taşır. İkinci alan geçerli ve nil olmayan UUID olmalıdır; örnek bunu outgoing olayın uygulama tarafından belirlenmiş kimliği olarak kullanır. Bu değer incoming delivery’nin canonical event ID’sinden ayrıdır. Aynı mantıksal outgoing olayı yeniden publish ederken kimliği koruyun; başka olaylar için yeniden kullanmayın.

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

Outbox hedefi domain.events exchange’i ve orders.persisted routing key’idir. Hedef binding’i ayrıca provision edin; publisher confirm için routable bir destination gerekir. Bu Consumer’a ait handler olmadan destination queue’sunu aynı Consumer topology’sine eklemek startup contract kontrolünü geçmez.

Concurrent duplicate ve retry#

Aynı handler/event anahtarını iki delivery eşzamanlı işlerse PostgreSQL’in non-blocking advisory lock kontrolü InProgress sonucunu retryable hataya çevirir. Bu durum normal queue retry/DLQ politikasını kullanır. Sık concurrent duplicate beklediğiniz queue’da retry_attempts = 0 seçiminin doğrudan dead-letter sonucunu değerlendirin.

Başarılı commit’ten sonra aynı event tekrar gelirse tamamlanmış inbox kaydı iş kodunu atlatır. PostgreSQL yolu MongoDB’deki transaction-label temelli bütün pipeline replay mekanizmasını kullanmaz. Uygulama retry’si ise ayrı bir broker delivery’sidir ve kendi scope’unu oluşturur.

Başlatmadan önce doğrulayın#

Normal startup migration çalıştırmaz. Lily, önceden uygulanmış schema sözleşmesini read-only preflight ile kontrol eder ve kabul edilen queue planından önce DB binding’ini hazırlar. Eksik migration veya schema drift olduğunda listener admission başlamaz. Bu, DI initializer’larının hiçbir ağ bağlantısı kurmadığı anlamına gelmez. Migration ve retention deployment sırasını açıklar.