MongoDB transaction’ları

MongoDB yolu, aynı DI DatabaseService üzerinden açılan session’ı bütün DB işlemlerine taşır. Transaction-capable replica set veya sharded deployment gerekir; standalone MongoDB uygun değildir.

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

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

MongoDB bağlantısını Lily’nin [database] bölümünde yapılandırın; bölümün adı [mongodb] değildir. İş collection’ı ve migrator aynı DatabaseService’i kullanmalıdır.

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

Bu top-level DB bölümü queue tablosunun dışındadır. config feature’ı örneğe eklenmiştir; ${secret:mongodb.primary_url} için secret resolver sağlayın. Çözülen URI transaction-capable replica set veya sharded deployment’a yönelmelidir. Diğer connection/TLS seçenekleri MongoDB entegrasyonunda açıklanır.

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

Tam mongodb.rs örneğini indirin.

İş verisi ve outgoing event#

transaction.operation_context() session’ı, deadline’ı ve cancellation’ı collection işlemine taşır. Normal DatabaseService::operation_context ile yeni context üretmek bu handler yazısını transaction’dan çıkarır. Context’i clone edip detached işe vermeyin; body tamamlanınca kullanım yetkisi sona erer.

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.

Örnek, collection yazısındaki hatayı öğretim amacıyla ORDER_WRITE_FAILED retryable hatasına çevirir. Bu bütün DB hatalarının geçici olduğu anlamına gelmez. Gerçek uygulamada validation, iş unique constraint’i ve kalıcı veri hatalarını transient bağlantı sorunlarından ayırarak error policy’nizi belirleyin.

MongoDB retry ve lease davranışı#

TransientTransactionError, middleware before/after, guard, extractor ve handler dahil bütün iş pipeline’ının yeni transaction/session ve yeni delivery scope ile tekrar çalışmasına yol açabilir. max_transaction_attempts ilk çalıştırmayı da sayar. Her denemede dış sistem çağrısı yapıyorsanız DB rollback bu çağrıyı geri alamaz.

UnknownTransactionCommitResult için aynı commit yeniden denenir; yalnız bu label nedeniyle body yeniden çalıştırılmaz. Commit reconciliation süresi hem MongoDB policy’si hem delivery’nin mevcut deadline’ı ile sınırlıdır. Sonuç hâlâ belirsizse lease expiry’ye bırakılır ve original delivery application retry/DLQ bütçesi tüketmeden doğrudan requeue edilir.

Aynı inbox anahtarındaki lease contention, mevcut absolute delivery bütçesi içinde bekler ve gerekirse original delivery’yi requeue eder. Commit sonrası lease release hatası advisory’dir: zaten Applied veya AlreadyCompleted olan sonucu iş başarısızlığına çevirmez. Policy değerlerini yapılandırma referansında inceleyin.

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.