الگوی Outbox و Inbox

مشکل: دو نوشتن، بدون تراکنش مشترک

مشتری سفارش ثبت می‌کند. سرویس سفارش دو کار انجام می‌دهد:

1. سفارش را در دیتابیس خودش ذخیره می‌کند.
2. پیام «سفارش ثبت شد» (OrderPlaced) را به Kafka یا RabbitMQ می‌فرستد. سرویس انبار با این پیام کالا را رزرو می‌کند.

کد ساده این‌طور است:

db.Orders.Add(order);
await db.SaveChangesAsync(ct);
await bus.PublishAsync(new OrderPlaced(order.Id, order.Lines), ct);

ظاهراً درست است. ولی دیتابیس و Broker دو سیستم جدا هستند. هیچ تراکنشی هر دو را با هم پوشش نمی‌دهد. به این مشکل Dual Write می‌گویند.

چه چیزی ممکن است خراب شود؟

1. برنامه بین دو خط می‌افتد. مثلاً هنگام انتشار نسخه جدید، Pod بسته می‌شود. سفارش هست، پیام نیست. انبار هیچ وقت کالا را رزرو نمی‌کند.
2. سرویس Broker چند ثانیه در دسترس نیست. فرستادن پیام خطا می‌دهد. سفارش قبلاً ذخیره شده و برنمی‌گردد.
3. ترتیب را عوض کنیم، مشکل برعکس می‌شود. اگر اول پیام را بفرستیم و بعد ذخیره شکست بخورد، انبار کالای سفارشی را رزرو می‌کند که اصلاً وجود ندارد.

ایده Outbox: پیام را هم در دیتابیس بنویس

دیتابیس فقط یک چیز را تضمین می‌کند: همه نوشتن‌های داخل یک تراکنش با هم انجام می‌شوند. پس پیام را هم در همان دیتابیس می‌نویسیم:

1. در یک تراکنش، هم سفارش را ذخیره می‌کنیم و هم یک ردیف در جدول OutboxMessages. این ردیف متن پیام است.
2. یک کار پس‌زمینه (Relay) ردیف‌های ارسال‌نشده را می‌خواند و به Broker می‌فرستد.
3. بعد از ارسال موفق، روی ردیف علامت «ارسال شد» می‌زند.

چرا این کار مشکل را حل می‌کند؟

1. اگر تراکنش موفق شد، پیام حتماً در جدول هست.
2. اگر برنامه افتاد، بعد از بالا آمدن، کار پس‌زمینه همان ردیف را پیدا می‌کند و می‌فرستد.
3. اگر Broker چند دقیقه قطع بود، پیام‌ها در جدول صبر می‌کنند. هیچ پیامی گم نمی‌شود.

ایده Inbox: پیام تکراری را بشناس

پیام تکراری فقط از Outbox نمی‌آید. Kafka و RabbitMQ هم ممکن است یک پیام را دوباره تحویل دهند. مثلاً مصرف‌کننده پیام را پردازش کرده، ولی قبل از تأیید (Ack یا Commit کردن Offset) افتاده است.

الگوی Inbox آینه Outbox در طرف گیرنده است:

1. هر پیام یک شناسه یکتا (MessageId) دارد.
2. سرویس انبار جدولی به اسم InboxMessages دارد. کلید اصلی آن، شناسه پیام است.
3. در یک تراکنش، هم شناسه پیام را در Inbox ثبت می‌کند و هم کالا را رزرو می‌کند.
4. اگر همین پیام دوباره برسد، ثبت شناسه با خطای کلید تکراری شکست می‌خورد. کل تراکنش برمی‌گردد و کالا دوباره رزرو نمی‌شود.
5. فقط بعد از ذخیره موفق، پیام را تأیید (Ack) می‌کنیم.

مثال زنده

گزینه‌ها را روشن یا خاموش کن و هر سناریو را اجرا کن. ببین در هر حالت چند رزرو در انبار ثبت می‌شود:

کد

ذخیره سفارش و پیام در یک تراکنش

جدول Outbox ساده است: شناسه پیام، نوع، متن و زمان ارسال.

public sealed class OutboxMessage
{
    public long Id { get; init; }                         // keeps the order of messages
    public Guid MessageId { get; init; } = Guid.NewGuid(); // consumers use it to find duplicates
    public required string Type { get; init; }
    public required string Payload { get; init; }
    public DateTimeOffset? SentAt { get; set; }
}

در EF Core، یک بار صدا زدن متد SaveChangesAsync خودش یک تراکنش است. پس هر دو ردیف با هم ذخیره می‌شوند:

app.MapPost("/orders", async (PlaceOrder cmd, ShopDbContext db, CancellationToken ct) =>
{
    var order = Order.Place(cmd.CustomerId, cmd.Lines);
    db.Orders.Add(order);
    db.OutboxMessages.Add(new OutboxMessage
    {
        Type = nameof(OrderPlaced),
        Payload = JsonSerializer.Serialize(new OrderPlaced(order.Id, order.Lines))
    });

    await db.SaveChangesAsync(ct); // one transaction: order + message
    return Results.Created($"/orders/{order.Id}", order.Id);
});

کار پس‌زمینه که پیام‌ها را می‌فرستد

public sealed class OutboxRelay(
    IServiceScopeFactory scopes, IMessagePublisher publisher, ILogger<OutboxRelay> logger)
    : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
        while (await timer.WaitForNextTickAsync(ct))
        {
            try { await SendBatchAsync(ct); }
            catch (Exception ex) when (!ct.IsCancellationRequested)
            {
                logger.LogError(ex, "Outbox batch failed. Will try again.");
            }
        }
    }

    private async Task SendBatchAsync(CancellationToken ct)
    {
        await using var scope = scopes.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<ShopDbContext>();

        var batch = await db.OutboxMessages
            .Where(m => m.SentAt == null)
            .OrderBy(m => m.Id)
            .Take(100)
            .ToListAsync(ct);

        foreach (var message in batch)
        {
            // IMessagePublisher is our own small wrapper around Kafka or RabbitMQ.
            await publisher.PublishAsync(message.MessageId, message.Type, message.Payload, ct);
            message.SentAt = DateTimeOffset.UtcNow;
            await db.SaveChangesAsync(ct);
        }
    }
}

اگر برنامه بین فرستادن و ذخیره علامت بیفتد، پیام دوباره فرستاده می‌شود. این طبیعی است. Inbox در طرف گیرنده آن را می‌گیرد.

مصرف‌کننده با Inbox

public sealed class OrderPlacedHandler(InventoryDbContext db)
{
    public async Task HandleAsync(Guid messageId, OrderPlaced message, CancellationToken ct)
    {
        db.InboxMessages.Add(new InboxMessage(messageId, DateTimeOffset.UtcNow));
        db.Reservations.Add(Reservation.For(message.OrderId, message.Lines));

        try
        {
            await db.SaveChangesAsync(ct); // inbox row + reservation, together
        }
        catch (DbUpdateException ex) when (IsDuplicateKey(ex))
        {
            // Same MessageId was processed before: do nothing.
        }
        // Ack the message only after this method returns.
    }

    private static bool IsDuplicateKey(DbUpdateException ex) =>
        ex.InnerException is PostgresException { SqlState: PostgresErrorCodes.UniqueViolation };
}

تشخیص خطای کلید تکراری به نوع دیتابیس بستگی دارد. این‌جا PostgreSQL با کتابخانه Npgsql است.

آماده: MassTransit

لازم نیست همه این‌ها را خودت بنویسی. کتابخانه MassTransit یک Outbox آماده برای EF Core دارد. همان جدول‌ها و کار پس‌زمینه را می‌سازد و در طرف گیرنده هم پیام تکراری را تشخیص می‌دهد:

builder.Services.AddMassTransit(x =>
{
    x.AddEntityFrameworkOutbox<ShopDbContext>(o =>
    {
        o.UsePostgres();
        o.UseBusOutbox(); // published messages go to the outbox table first
    });

    x.UsingRabbitMq((context, cfg) => cfg.ConfigureEndpoints(context));
});

قانون‌های مهم

1. پیام و داده در یک تراکنش. اگر ردیف Outbox در تراکنش دیگری ذخیره شود، همان مشکل Dual Write برمی‌گردد.
2. انتظار پیام تکراری را داشته باش. Outbox تحویل حداقل یک بار می‌دهد. پس هر مصرف‌کننده باید Idempotent باشد.
3. هر پیام یک شناسه یکتا داشته باشد. این شناسه در همه تلاش‌های ارسال ثابت است. گیرنده با آن تکراری‌ها را می‌شناسد.
4. تأیید پیام بعد از ذخیره. اگر قبل از پردازش Ack کنی و برنامه بیفتد، پیام گم می‌شود.
5. روی چند Pod مراقب باش. اگر چند نسخه از کار پس‌زمینه یک ردیف را با هم بخوانند، پیام چند بار می‌رود. یا فقط یک نسخه را اجرا کن، یا ردیف‌ها را قفل کن (مثلاً با دستور FOR UPDATE SKIP LOCKED در PostgreSQL). اگر ترتیب پیام‌ها مهم است، این انتخاب را با دقت انجام بده.
6. جدول‌ها را تمیز کن. ردیف‌های قدیمی Outbox و Inbox را بعد از مدتی پاک کن. جدول Inbox باید حداقل به اندازه بیشترین زمان ممکن برای رسیدن پیام تکراری نگه داشته شود.
7. کار بیرون از دیتابیس را جدا ببین. Inbox فقط تغییرات همان دیتابیس را دقیقاً یک بار می‌کند. اگر پیامک بفرستی و قبل از ذخیره بیفتی، پیامک دوباره می‌رود. برای این کارها، یک شناسه یکتا به سرویس بیرونی بده، اگر پشتیبانی می‌کند.

اشتباه‌های رایج

چه وقت Outbox و Inbox؟