Как гарантировать откат передачи сообщения при падении одного микросервиса на пути передачи?

Гарантия отката (Rollback) при распределенной передаче сообщений

В распределенной системе, где сообщение проходит через цепочку микросервисов, обеспечение транзакционности и отката при падении одного из участников — комплексная задача, поскольку классические ACID-транзакции здесь неприменимы. Вместо этого используются паттерны управления сагами (Saga Pattern) и компенсирующие транзакции.

Ключевые стратегии

1. Паттерн "Сага"

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

• Оркестрируемые саги: Центральный координатор (оркестратор) управляет потоком и откатом.
• Хореографические саги: Каждый сервис самостоятельно запускает следующий шаг или компенсацию через события.

Пример кода оркестратора на Go:

// Пример структуры оркестратора
type TransferOrchestrator struct {
    steps        []Step
    compensation []Compensation
}

func (o *TransferOrchestrator) Execute(ctx context.Context, transferID string) error {
    for i, step := range o.steps {
        if err := step.Execute(ctx, transferID); err != nil {
            // Запускаем компенсации в обратном порядке
            for j := i - 1; j >= 0; j-- {
                if errComp := o.compensation[j].Execute(ctx, transferID); errComp != nil {
                    // Логируем ошибку компенсации, но продолжаем откат
                    log.Printf("Compensation failed for step %d: %v", j, errComp)
                }
            }
            return fmt.Errorf("step %d failed: %w", i, err)
        }
    }
    return nil
}

2. Компенсирующие транзакции

Каждый сервис должен реализовать идемпотентную операцию отката, которая отменяет эффект основной операции. Например, если сервис списал деньги, компенсация — их возврат.

// Пример компенсирующей транзакции для сервиса платежей
type PaymentService struct {
    repo PaymentRepository
}

func (s *PaymentService) CompensatePayment(ctx context.Context, paymentID string) error {
    payment, err := s.repo.GetByID(paymentID)
    if err != nil {
        return err
    }
    // Идемпотентность: проверяем, не был ли уже откат
    if payment.Status == "compensated" {
        return nil
    }
    // Возвращаем средства
    payment.Status = "compensated"
    return s.repo.Update(payment)
}

3. Использование очередей сообщений с подтверждениями

Очереди вроде Apache Kafka или RabbitMQ позволяют гарантировать доставку через подтверждения (acknowledgments) и механизмы повтора. Например, можно не подтверждать сообщение, пока все шаги не завершены.

4. Паттерн "Повторитель-распределитель (Retry-Dispatcher)"

Сервис сохраняет сообщение в локальную БД перед обработкой и удаляет только после успеха. При падении и перезапуске, необработанные сообщения из БД снова отправляются в очередь.

// Пример сохранения сообщения перед обработкой
func (s *Service) ProcessMessage(msg Message) error {
    // 1. Сохраняем в БД с статусом "processing"
    if err := s.repo.SaveMessage(msg, "processing"); err != nil {
        return err
    }
    // 2. Выполняем бизнес-логику
    if err := s.handleBusinessLogic(msg); err != nil {
        s.repo.UpdateStatus(msg.ID, "failed")
        return err
    }
    // 3. Обновляем статус на "completed"
    return s.repo.UpdateStatus(msg.ID, "completed")
}

Практические шаги реализации

1. Идемпотентность операций: Все операции, включая компенсации, должны быть идемпотентными, чтобы повторные вызовы не вызывали побочных эффектов.
2. Логирование состояний: Каждый сервис должен хранить состояние обработки сообщения (например, в БД), чтобы при перезапуске продолжить с точки сбоя.
3. Таймауты и мониторинг: Устанавливайте таймауты на операции и используйте Circuit Breaker для избежания каскадных сбоев.
4. Асинхронная связь: Используйте асинхронные сообщения через брокер, чтобы избежать блокировок и повысить отказоустойчивость.

Пример полного flow с откатом

1. Сервис A начинает сагу, публикует событие TransferStarted.
2. Сервис B получает событие, списывает деньги, публикует MoneyDebited.
3. Сервис C падает при попытке зачислить деньги.
4. Оркестратор обнаруживает таймаут, публикует CompensateDebit для B.
5. Сервис B выполняет компенсацию, возвращает деньги.

Заключение: Гарантия отката в распределенных системах требует сочетания паттернов саг, компенсирующих транзакций, идемпотентности и надежных механизмов доставки сообщений. На Go это реализуется через структурированные обработчики, использование транзакций в БД для хранения состояния и асинхронные коммуникации через брокеры. Важно также предусмотреть мониторинг и ручные интервенции для случаев, когда автоматический откат невозможен.