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

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

Важно знать

• Compensation — отмена эффекта предыдущего шага саги средствами бизнес-логики, не технический rollback СУБД.
• Каждая command, участвующая в саге, имеет парную compensation-команду: charge_payment ↔ refund_payment, reserve_inventory ↔ release_inventory, create_order ↔ cancel_order.
• Compensation идемпотентна — orchestrator повторит её при retry. Повторный refund одной и той же транзакции = no-op: проверяй status до действия.
• Semantic, не технический. Refund — новая строка в БД и новая транзакция у провайдера. Деньги уже ушли, их возвращают, а не «не списывают задним числом».
• Audit trail обязателен — status=REFUNDED, refund_id, refunded_at, refund_reason. DELETE теряет историю и ломает внешние ключи.
• Failure compensation — отдельный сценарий. Если сам refund упал — COMPENSATION_FAILED + DLQ + алерт, не log.error и тишина.
• В Python транзакция — это async with session.begin() (UoW через SQLAlchemy AsyncSession), не @Transactional; всё остальное — та же архитектура.

Сага без compensation — оптимистичная цепочка. Payment-сервис отвечает «ок» на 99.9% запросов, легко забыть про 0.1%: остальные шаги уже прошли, а payment нужно откатить. Compensation — единственный способ корректно завершить in-flight сагу при сбое.

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

R-DIST-COMP-1: каждая forward-команда, меняющая состояние, имеет compensation-команду в том же сервисе.

Compensation — отдельный use case с отдельным handler. Orchestrator саги вызывает его как любую другую команду.

from uuid import UUID
from sqlalchemy.ext.asyncio import AsyncSession
from app.domain.payment import Payment, PaymentStatus
from app.ports.payment_provider import PaymentProviderPort
from app.ports.outbox import OutboxPort
from app.domain.events import PaymentRefundedEvent


class RefundPaymentHandler:
    def __init__(
        self,
        session: AsyncSession,
        payment_provider: PaymentProviderPort,
        outbox: OutboxPort,
    ) -> None:
        self._session = session
        self._payment_provider = payment_provider
        self._outbox = outbox

    async def handle(self, payment_id: UUID, reason: str) -> Payment:
        async with self._session.begin():
            payment = await self._session.get(
                Payment, payment_id, with_for_update=True
            )
            if payment is None:
                raise PaymentNotFoundError(payment_id)

            if payment.status == PaymentStatus.REFUNDED:
                return payment

            if payment.status != PaymentStatus.CHARGED:
                raise PaymentNotRefundableError(payment_id, payment.status)

            refund_id = await self._payment_provider.refund(
                payment.external_id, payment.amount
            )
            payment.mark_refunded(refund_id, reason)
            await self._outbox.publish(
                PaymentRefundedEvent(payment_id=payment_id, refund_id=refund_id)
            )
        return payment

with_for_update=True на SELECT — обязателен: два параллельных retry orchestrator'а не должны оба пройти status != REFUNDED и отправить двойной refund.

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

R-DIST-COMP-2: orchestrator может вызвать refund несколько раз — retry на network timeout, рестарт процесса, TaskIQ/Temporal повтор. Compensation handler возвращает тот же результат при повторе.

В примере выше if payment.status == PaymentStatus.REFUNDED: return payment делает повторный вызов no-op. Дополнительно payment_provider.refund идемпотентен на уровне провайдера через Idempotency-Key (см. Idempotency).

Без идемпотентности: orchestrator повторяет refund, провайдер делает refund дважды, клиент получает удвоенный возврат — поддержка неделю разбирается.

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

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

Forward:      charge_payment  → CHARGED   → деньги у банка
Compensation: refund_payment  → REFUNDED  → деньги вернулись клиенту

В БД — две строки:

payment(id=42, status=REFUNDED, external_id=..., amount=990_00)
refund(id=7,   payment_id=42, external_refund_id=..., amount=990_00, reason='INVENTORY_UNAVAILABLE')

Не одна запись с status=CHARGED → DELETE. Деньги двигались туда и обратно — в БД должны быть обе операции.

Аналогично для инвентаря:

async def release_inventory(order_id: UUID, session: AsyncSession) -> None:
    async with session.begin():
        reservation = await session.execute(
            select(Reservation)
            .where(Reservation.order_id == order_id)
            .with_for_update()
        )
        row = reservation.scalar_one_or_none()
        if row is None or row.status == ReservationStatus.RELEASED:
            return
        row.status = ReservationStatus.RELEASED
        row.released_at = datetime.utcnow()

Не session.delete(row) — чтобы история резервирования осталась и была видна в audit.

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

R-DIST-COMP-4: compensation меняет статус и добавляет ссылку на исходную операцию, не теряет данные.

class Payment(Base):
    __tablename__ = "payment"

    id: Mapped[UUID] = mapped_column(primary_key=True)
    order_id: Mapped[UUID] = mapped_column(nullable=False)
    status: Mapped[PaymentStatus] = mapped_column(nullable=False)
    external_id: Mapped[str] = mapped_column(nullable=False)
    amount: Mapped[int] = mapped_column(nullable=False)
    refund_id: Mapped[str | None] = mapped_column(nullable=True)
    refunded_at: Mapped[datetime | None] = mapped_column(nullable=True)
    refund_reason: Mapped[str | None] = mapped_column(nullable=True)

    def mark_refunded(self, refund_id: str, reason: str) -> None:
        self.status = PaymentStatus.REFUNDED
        self.refund_id = refund_id
        self.refunded_at = datetime.utcnow()
        self.refund_reason = reason

После compensation SQL-запрос даёт полную картину:

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

-- id  | status   | refund_id | refunded_at         | refund_reason
-- 42  | REFUNDED | ref-007   | 2026-06-18 14:30:00 | INVENTORY_UNAVAILABLE

Это отвечает на «почему клиент получил возврат» и «какая сага инициировала refund» без расследования в Kafka-логах.

Failure compensation — DLQ

Compensation сам может упасть: endpoint провайдера недоступен, сетевой сбой, таймаут. Простой retry в orchestrator'е решает большинство случаев, но не все.

from app.domain.saga import SagaStatus
from app.ports.alerting import AlertingPort


class OrderSagaOrchestrator:
    async def _compensate(
        self,
        saga_id: UUID,
        order_id: UUID | None,
        payment_id: UUID | None,
    ) -> None:
        await self._saga_repo.update_status(saga_id, SagaStatus.COMPENSATING)
        try:
            if payment_id is not None:
                await self._payment_service.refund(payment_id, reason="SAGA_FAILED")
            if order_id is not None:
                await self._order_service.cancel(order_id, reason="SAGA_FAILED")
            await self._saga_repo.update_status(saga_id, SagaStatus.FAILED)
        except Exception as exc:
            await self._saga_repo.update_status(
                saga_id, SagaStatus.COMPENSATION_FAILED
            )
            await self._compensation_dlq.enqueue(saga_id, str(exc))
            await self._alerting.notify_ops(saga_id, exc)

COMPENSATION_FAILED — терминальный статус саги. Деньги «висят», ops связывается с клиентом и чинит вручную. «Потерять и забыть» недопустимо для денег.

Если используется Temporal — compensation-шаг реализуется как отдельная Activity с собственным retry_policy:

from temporalio import activity, workflow
from temporalio.common import RetryPolicy
from datetime import timedelta


@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, input: OrderSagaInput) -> None:
        try:
            await workflow.execute_activity(
                charge_payment,
                input.payment_data,
                start_to_close_timeout=timedelta(seconds=30),
            )
            await workflow.execute_activity(
                reserve_inventory,
                input.inventory_data,
                start_to_close_timeout=timedelta(seconds=30),
            )
        except Exception:
            await workflow.execute_activity(
                refund_payment,
                input.payment_data,
                start_to_close_timeout=timedelta(seconds=60),
                retry_policy=RetryPolicy(maximum_attempts=10),
            )
            raise

retry_policy на compensation-активность гарантирует повтор до исчерпания лимита; после — Temporal переводит workflow в failed с полным trail'ом.

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

Куда дальше

• Saga — где вызывается compensation в orchestrator'е, структура OrderSagaOrchestrator.
• Idempotency — compensation идемпотентна через status-check и Idempotency-Key.
• Outbox + Inbox — события compensation публикуются через outbox, не прямым producer.send.
• Eventual consistency — read-модели после compensation и bounded staleness.
• Distributed transactions — почему 2PC не альтернатива саге с compensation.
• When to use — критерии выбора между локальным UoW и распределёнными паттернами.