Bios.Outbox

PostgreSQL üzerinde transactional outbox. Bir iş, onu doğuran veritabanı transaction'ının içinde kuyruğa yazılır; transaction geri alınırsa iş de yazılmamış olur. Ayrı bir işçi kuyruğu FOR UPDATE SKIP LOCKED ile boşaltır, her kuyruk kendi eşzamanlılık sınırıyla çalışır.

Biten iş silinir. Deneme hakkı tükenen iş dead durumuna düşer ve elle incelenir. Her başarısız deneme bir sonrakini üstel gecikmeyle erteler.

Tablo

Kütüphane şu tabloyu bekler; kendi göçüne ekle:

create table outbox_jobs (
    id         bigint generated always as identity primary key,
    queue      text not null,
    kind       text not null,
    payload    jsonb not null default '{}',
    status     text not null default 'queued'
               check (status in ('queued', 'running', 'dead')),
    attempts   int not null default 0,
    run_after  timestamptz not null default now(),
    created_at timestamptz not null default now(),
    started_at timestamptz,
    last_error text
);

create index outbox_jobs_ready_idx
    on outbox_jobs (queue, run_after, id) where status = 'queued';

Rol ayrımı olan bir sistemde kuyruğa yazan role select, insert, işçi rolüne select, update, delete verilir.

Bağlama

Üç adım: depo bir kez, her iş türü için bir işleyici, her kuyruk için bir işçi.

services.AddBiosOutbox(dataSource);                 // NpgsqlDataSource host'un
services.AddBiosOutboxHandler<TransferItemHandler>();
services.AddBiosOutboxQueue(new OutboxQueueOptions { Name = "transfer", MaxConcurrency = 50 });

Host'un paylaşacak bir NpgsqlDataSource'u yoksa bağlantı dizesi de verilebilir; bu durumda depo kendi veri kaynağını açar ve host kapanırken onu kendisi kapatır:

services.AddBiosOutbox(connectionString);

İşleyici bir iş türünü karşılar; Kind kuyruğa yazarken verilen türle birebir eşleşmelidir:

public sealed class TransferItemHandler : IOutboxHandler
{
    public string Kind => "transfer.item";

    public async Task HandleAsync(OutboxJob job, CancellationToken ct)
    {
        // job.Payload -- kuyruğa yazılan JSON metni
    }
}

Kuyruğa yazmak IOutbox ile olur. Asıl kullanım, işi doğuran veritabanı işleminin kendi transaction'ı içindedir:

await outbox.EnqueueAsync(connection, transaction, "transfer", "transfer.item", payloadJson, ct);

İş tek başına duruyorsa, yani bir yazmaya bağlı değilse, kısa biçim kendi bağlantısını açar:

await outbox.EnqueueAsync("transfer", "transfer.item", payloadJson, ct);

Farklı maliyet sınıfları farklı kuyruklara konur; bir kuyruğun yavaşlığı ötekini bekletmez.

Yapılandırma

Bu paket appsettings okumaz; bir kuyruğun bütün ayarı AddBiosOutboxQueue'ya verilen OutboxQueueOptions kaydıdır. Uyarlanabilir hiçbir şey yoktur: eşzamanlılık senin verdiğin sayıdır.

services.AddBiosOutboxQueue(new OutboxQueueOptions
{
    Name = "transfer",
    MaxConcurrency = 50,
    BatchSize = 200,
});

services.AddBiosOutboxQueue(new OutboxQueueOptions
{
    Name = "parse",          // ağır iş, kendi kuyruğunda
    MaxConcurrency = 4,
});

Dikkat

İşleyici aynı işi iki kez görebilir. İş, satır silinmeden hemen önce çöken bir işçiden sonra yeniden kuyruğa girer. "Bunu zaten yapmış mıydım" sorusunun cevabı işleyiciye aittir.

İşçi host'u düşürmez. Veritabanına ulaşılamıyorsa işçi büyüyen aralıklarla yeniden dener ve pes etmez; host ayakta kalır, HTTP yüzeyi cevap vermeyi sürdürür. Veritabanı döndüğünde kuyruk kaldığı yerden devam eder.

Tablo kendini kurar, ama sessizce vazgeçer. Tablo yoksa işçi ilk açılışta onu oluşturur. Tablo zaten varsa hiçbir şey yapmaz — göç ile yönetilen bir şemada DDL yetkisi olmayan bir rolle çalışmak bu yüzden sorun değildir.

Türler metindir, eşleşmesi senin işin. Kuyruğa yazarken verilen kind ile işleyicinin Kind değeri birebir tutmazsa iş sahipsiz kalır: her denemede no_handler hatasıyla düşer ve MaxAttempts dolunca dead olur.