Compensation в saga — Go: semantic state-change, идемпотентность, DLQ

Опирается на правила: R-DIST-COMP-1 … R-DIST-COMP-4 и R-DIST-COMP-X1 … R-DIST-COMP-X3 из Distributed Patterns — раздел 6. Compensation.

Важно знать

• Compensation — отмена эффекта шага саги, не технический rollback средствами pgx.
• Каждая command в саге имеет парную compensation-команду: ReservePayment ↔ RefundPayment, ReserveInventory ↔ ReleaseInventory, CreateOrder ↔ CancelOrder.
• Go не даёт аннотаций — транзакция явная: pgx.Tx, пробрасываемый в queries.WithTx(tx); никакого магического rollback нет.
• Compensation идемпотентна: перед действием проверяется статус (status == "refunded" → return nil); saga может повторить её сколько угодно раз при retry.
• Semantic, не технический. Refund — новая запись в БД и новый вызов платёжного провайдера, не «откат» pgx.Tx.
• Audit trail обязателен: compensation меняет статус на refunded со ссылкой на оригинал; DELETE запрещён.
• Failure compensation — терминальный сценарий: refund упал → status = "compensation_failed" → DLQ-топик → алерт; silent fail недопустим.
• Compensation-команды публикуются через outbox в той же pgx.Tx, в которой обновляется статус саги.

Сага без compensation — оптимистичная цепочка, а не транзакционный паттерн. Когда payment-service ответил «ок» и пошёл шаг inventory, а тот упал — без compensation у OrderSagaOrchestrator нет способа корректно завершить in-flight сагу. Деньги зарезервированы, заказ ни в каком состоянии. Compensation — единственный инструмент обратного хода.

Парная compensation-команда

R-DIST-COMP-1: каждый forward-шаг саги имеет compensation-команду в том же сервисе.

Compensation — отдельный use case с отдельным HTTP-обработчиком у каждого сервиса. OrderSagaOrchestrator вызывает его через outbox точно так же, как любую другую команду.

// internal/saga/order_saga.go

func (o *OrderSagaOrchestrator) OnInventoryFailed(
    ctx context.Context, tx pgx.Tx, event InventoryFailedEvent,
) error {
    qtx := o.queries.WithTx(tx)

    if err := qtx.UpdateOrderSagaStatus(ctx, db.UpdateOrderSagaStatusParams{
        SagaID:      event.SagaID,
        Status:      "compensating",
        CurrentStep: "refund_payment",
    }); err != nil {
        return fmt.Errorf("update saga to compensating: %w", err)
    }

    return o.outbox.Write(ctx, tx, outbox.Message{
        SagaID:  event.SagaID,
        Topic:   "payment.commands",
        Payload: RefundPaymentCommand{
            SagaID:          event.SagaID,
            OriginalOrderID: event.OrderID,
            Reason:          "inventory_unavailable",
        },
    })
}

Статус саги переходит в "compensating" атомарно с записью compensation-команды в outbox — в одной pgx.Tx. Если сервис упадёт до коммита, оба действия откатятся и saga возобновится с предыдущего шага.

Идемпотентность compensation

R-DIST-COMP-2: orchestrator может повторить compensation при retry на network timeout или после рестарта. Handler compensation обязан возвращать тот же результат при повторе.

// core/payment/refund_handler.go

type RefundPaymentHandler struct {
    queries         *db.Queries
    pool            *pgxpool.Pool
    paymentProvider PaymentProviderPort
    outbox          OutboxWriter
}

func (h *RefundPaymentHandler) Handle(ctx context.Context, cmd RefundPaymentCommand) error {
    tx, err := h.pool.Begin(ctx)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback(ctx)

    qtx := h.queries.WithTx(tx)

    payment, err := qtx.GetPaymentForUpdate(ctx, cmd.OriginalOrderID)
    if err != nil {
        return fmt.Errorf("get payment: %w", err)
    }

    if payment.Status == "refunded" {
        return nil
    }
    if payment.Status != "charged" {
        return &PaymentNotRefundableError{PaymentID: payment.ID, Status: payment.Status}
    }

    refundID, err := h.paymentProvider.Refund(ctx, payment.ExternalID, payment.Amount, cmd.SagaID.String())
    if err != nil {
        return fmt.Errorf("provider refund: %w", err)
    }

    if err := qtx.MarkPaymentRefunded(ctx, db.MarkPaymentRefundedParams{
        PaymentID: payment.ID,
        RefundID:  refundID,
        Reason:    cmd.Reason,
    }); err != nil {
        return fmt.Errorf("mark payment refunded: %w", err)
    }

    if err := h.outbox.Write(ctx, tx, outbox.Message{
        SagaID:  cmd.SagaID,
        Topic:   "payment.events",
        Payload: PaymentRefundedEvent{SagaID: cmd.SagaID, OrderID: cmd.OriginalOrderID, RefundID: refundID},
    }); err != nil {
        return fmt.Errorf("write outbox: %w", err)
    }

    return tx.Commit(ctx)
}

Идемпотентность обеспечивается двойной защитой: проверкой статуса до действия (status == "refunded" → return nil) и тем, что paymentProvider.Refund получает saga_id как Idempotency-Key — провайдер вернёт тот же refundID при повторе.

Semantic compensation, не технический rollback

R-DIST-COMP-3: payment compensation — это refund (новая транзакция), не откат pgx.Tx.

Forward:      ReservePayment  → status = "charged"   → деньги у банка
Compensation: RefundPayment   → status = "refunded"  → деньги вернулись

В БД — две сущности:

SELECT id, status, refund_id, refunded_at, refund_reason
FROM payment
WHERE order_id = 'ord-7829';

-- id           | status   | refund_id     | refunded_at          | refund_reason
-- pay-3341     | refunded | ref-9912      | 2026-06-20 10:15:00  | inventory_unavailable

Отдельно хранится факт резервирования инвентаря:

Forward:      ReserveInventory  → INSERT reservation(status = "active")
Compensation: ReleaseInventory  → UPDATE reservation SET status = "released", released_at = now()

Не DELETE FROM reservation — нужно знать, что резервирование было, когда и почему отменилось.

Audit trail обязателен

R-DIST-COMP-4: compensation меняет статус с reference к исходной операции. DELETE и UPDATE с потерей истории запрещены.

// db/query/payment.sql → sqlc-generated

// MarkPaymentRefunded — атомарно записывает refund_id, reason, refunded_at
func (q *Queries) MarkPaymentRefunded(ctx context.Context, arg MarkPaymentRefundedParams) error {
    const query = `
        UPDATE payment
        SET status       = 'refunded',
            refund_id    = $2,
            refund_reason = $3,
            refunded_at  = now()
        WHERE id = $1
          AND status = 'charged'
    `
    _, err := q.db.Exec(ctx, query, arg.PaymentID, arg.RefundID, arg.Reason)
    return err
}

Условие AND status = 'charged' — дополнительная защита от гонки: если параллельный поток уже сделал refund, UPDATE вернёт 0 строк, а handler заметит это через RowsAffected().

Для заказа compensation — смена статуса, не удаление:

// db/query/order.sql → sqlc-generated

// CancelOrder — compensation для CreateOrder
func (q *Queries) CancelOrder(ctx context.Context, arg CancelOrderParams) error {
    const query = `
        UPDATE sber_order
        SET status        = 'cancelled',
            cancel_reason = $2,
            cancelled_at  = now()
        WHERE id = $1
          AND status NOT IN ('cancelled', 'completed')
    `
    _, err := q.db.Exec(ctx, query, arg.OrderID, arg.Reason)
    return err
}

Saga завершает compensation

После подтверждения refund orchestrator переводит сагу в терминальный статус "failed" и меняет статус заказа:

// internal/saga/order_saga.go

func (o *OrderSagaOrchestrator) OnPaymentRefunded(
    ctx context.Context, tx pgx.Tx, event PaymentRefundedEvent,
) error {
    qtx := o.queries.WithTx(tx)

    if err := qtx.UpdateOrderSagaStatus(ctx, db.UpdateOrderSagaStatusParams{
        SagaID:      event.SagaID,
        Status:      "failed",
        CurrentStep: "terminal",
    }); err != nil {
        return fmt.Errorf("mark saga failed after refund: %w", err)
    }

    return qtx.CancelOrder(ctx, db.CancelOrderParams{
        OrderID: event.OrderID,
        Reason:  "payment_refunded",
    })
}

Failure compensation — DLQ

Compensation сам может упасть: сеть лежит, платёжный провайдер недоступен, БД перегружена. Простой retry из orchestrator решает большинство случаев. Но если retry исчерпан — нужен DLQ.

// adapters/in/kafka/saga_consumer.go

func (c *SagaConsumer) handleWithDLQ(ctx context.Context, msg kafka.Message) {
    if err := c.handle(ctx, msg); err != nil {
        sagaID := headerValue(msg.Headers, "saga_id")
        slog.ErrorContext(ctx, "compensation failed, routing to DLQ",
            "saga_id", sagaID,
            "topic", msg.Topic,
            "error", err,
        )

        tx, txErr := c.pool.Begin(ctx)
        if txErr == nil {
            defer tx.Rollback(ctx)
            _ = c.queries.WithTx(tx).UpdateOrderSagaStatus(ctx, db.UpdateOrderSagaStatusParams{
                SagaID:      mustParseUUID(sagaID),
                Status:      "compensation_failed",
                CurrentStep: "terminal",
            })
            _ = tx.Commit(ctx)
        }

        _ = c.dlqWriter.WriteMessages(ctx, toDLQMessage(msg, err))
    }
}

"compensation_failed" — терминальный статус саги. Деньги «висят», нужен manual review. DLQ-топик мониторится алертом: любое сообщение в payment.compensation.dlq → PagerDuty.

Без DLQ:

• "compensating" навечно в saga_order.
• Никто не знает. Клиент ждёт возврат. Деньги заморожены у провайдера.
• Support разбирает инцидент через неделю по логам.

Что запрещено

Куда дальше

• Saga — оркестрация vs хореография — где вызывается compensation в OrderSagaOrchestrator.
• Idempotency — compensation идемпотентна через статус-проверку и Idempotency-Key у провайдера.
• Outbox + Inbox — compensation-команды публикуются через outbox в той же pgx.Tx.
• Eventual consistency — состояние после compensation распространяется асинхронно.
• Distributed transactions — что не делать — почему tx.Rollback не замена compensation.
• Когда нужны распределённые паттерны — compensation актуальна только при cross-service операции.