Что делать, если транзакция падает при отправке сообщения и данные теряются?

В распределенных системах прямая отправка сообщения в очередь (например, RabbitMQ, Kafka) внутри бизнес-транзакции создает проблему: если транзакция в базе данных откатывается, сообщение уже могло быть отправлено и обработано потребителем, что приводит к несогласованности данных. Решением является паттерн "Transactional Outbox" (или "Исходящий почтовый ящик").

Как работает паттерн Outbox

Вместо немедленной отправки сообщения в очередь, приложение выполняет следующие шаги в рамках одной транзакции с базой данных:

1. Обновляет состояние бизнес-сущности (например, меняет статус заказа на "оплачен").
2. Вставляет запись в специальную таблицу outbox в той же базе данных. Эта запись содержит данные события (сообщения) в формате JSON или другом сериализованном виде.

Только после успешного коммита транзакции, событие считается сохраненным. Отправкой сообщения в очередь занимается отдельный фоновый процесс (реле или издатель), который периодически опрашивает таблицу outbox на наличие новых записей, отправляет их в очередь и удаляет или помечает как отправленные.

Пример кода (упрощенный)

// 1. Внутри сервисного слоя
public async Task ProcessOrderAsync(Order order) {
    using var transaction = await db.Database.BeginTransactionAsync();
    try {
        // Бизнес-логика: обновление заказа
        order.Status = OrderStatus.Paid;
        db.Orders.Update(order);

        // 2. Сохранение события в Outbox ВМЕСТО прямой отправки
        var outboxMessage = new OutboxMessage {
            Id = Guid.NewGuid(),
            EventType = "OrderPaid",
            Payload = JsonSerializer.Serialize(new { OrderId = order.Id }),
            CreatedAt = DateTime.UtcNow
        };
        db.OutboxMessages.Add(outboxMessage);

        // ВСЕ изменения в одной транзакции
        await db.SaveChangesAsync();
        await transaction.CommitAsync(); // Только после коммита событие сохраняется
    } catch {
        await transaction.RollbackAsync();
        throw;
    }
    // 3. Отдельный процесс (Publisher) позже заберет сообщение из Outbox и отправит в очередь
}

// 4. Фоновый процесс-издатель (запускается, например, по таймеру)
public async Task PublishOutboxMessagesAsync() {
    var messages = await db.OutboxMessages
        .Where(m => !m.Processed)
        .OrderBy(m => m.CreatedAt)
        .Take(100)
        .ToListAsync();

    foreach (var msg in messages) {
        try {
            await messageBus.PublishAsync(msg.EventType, msg.Payload);
            msg.Processed = true; // Помечаем как отправленное
        } catch (Exception ex) {
            // Логируем ошибку, можно повторить позже
        }
    }
    await db.SaveChangesAsync(); // Сохраняем статус отправки
}

Дополнительные меры надежности

• Идемпотентность потребителя: Получатель сообщения должен обрабатывать его идемпотентно (повторная обработка того же события не должна менять результат). Это можно достичь, сохраняя ID обработанных сообщений или используя семантические идентификаторы (например, OrderId + статус).
• Гарантированная доставка: Процесс-издатель должен иметь механизм повторных попыток (retry) при временных сбоях очереди.
• Дедупликация в очереди: Некоторые брокеры (Kafka) поддерживают идемпотентных продюсеров, предотвращая дубли на этапе отправки.

Вывод: Паттерн Transactional Outbox — это стандартный подход для гарантированной отправки сообщений в асинхронных сценариях, где критична согласованность данных между базой и очередью. Его стоит применять в микросервисной архитектуре, при интеграции с системами событий (Event-Driven), или в любом случае, когда потеря или дублирование сообщения недопустимы.