> Как использовать паттерн outbox для обработки задержек записи в базу и обеспечения отказоустойчивости (Go)

Паттерн Outbox решает проблему атомарности между записью в базу данных и отправкой сообщения (или вызовом внешнего сервиса). В Go это реализуется так:

1. Основная идея: Вместо прямой отправки сообщения в брокер, мы сначала сохраняем "исходящее сообщение" в той же транзакции, что и бизнес-данные. Отдельный процесс (outbox relay) читает эти сообщения и отправляет их.

2. Пример реализации на Go:

GO
type Order struct {
    ID     int64
    Status string
}

type OutboxMessage struct {
    ID        int64
    Topic     string
    Payload   []byte
    CreatedAt time.Time
}

// Сохранение заказа и outbox-сообщения в одной транзакции
func CreateOrder(ctx context.Context, db *sql.DB, order Order) error {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // Бизнес-логика
    _, err = tx.ExecContext(ctx, "INSERT INTO orders (id, status) VALUES ($1, $2)", order.ID, order.Status)
    if err != nil {
        return err
    }

    // Сообщение в outbox
    payload, _ := json.Marshal(order)
    _, err = tx.ExecContext(ctx, 
        "INSERT INTO outbox (topic, payload, created_at) VALUES ($1, $2, $3)",
        "order_created", payload, time.Now())
    if err != nil {
        return err
    }

    return tx.Commit()
}

// Outbox relay - отдельная горутина
func OutboxRelay(ctx context.Context, db *sql.DB, producer MessageProducer) {
    for {
        select {
        case <-ctx.Done():
            return
        default:
            // Читаем неотправленные сообщения
            rows, err := db.QueryContext(ctx, 
                "SELECT id, topic, payload FROM outbox WHERE sent_at IS NULL ORDER BY id LIMIT 100")
            if err != nil {
                time.Sleep(time.Second)
                continue
            }

            for rows.Next() {
                var msg OutboxMessage
                rows.Scan(&msg.ID, &msg.Topic, &msg.Payload)

                // Отправляем в брокер
                err := producer.Send(msg.Topic, msg.Payload)
                if err != nil {
                    log.Printf("Failed to send message %d: %v", msg.ID, err)
                    continue
                }

                // Помечаем как отправленное
                db.ExecContext(ctx, "UPDATE outbox SET sent_at = NOW() WHERE id = $1", msg.ID)
            }
            rows.Close()

            time.Sleep(100 * time.Millisecond) // Пауза между итерациями
        }
    }
}

3. Обработка задержек и отказоустойчивость:

   • Задержки записи: Outbox relay работает асинхронно, не блокируя основной поток. Если БД медленная, сообщения накапливаются в таблице и отправляются позже.
   • Отказоустойчивость: При падении сервиса после коммита транзакции, но до отправки сообщения, relay при перезапуске подхватит неотправленные сообщения (sent_at IS NULL).
   • Идемпотентность: Добавьте уникальный ключ (например, event_id) в outbox, чтобы избежать дубликатов при повторной отправке.

4. Дополнительные улучшения:

   • Используйте SELECT ... FOR UPDATE SKIP LOCKED для параллельной обработки несколькими relay.
   • Добавьте TTL для старых сообщений, чтобы не засорять таблицу.
   • Для высоких нагрузок используйте batch-отправку сообщений.

Этот паттерн гарантирует, что каждое бизнес-событие будет доставлено хотя бы один раз (at-least-once), даже при сбоях.