Compensation в saga — NestJS: 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 средствами TypeORM.
• Каждая command в саге имеет парную compensation-команду: RequestPaymentCommand ↔ RefundPaymentCommand, ReserveInventoryCommand ↔ ReleaseInventoryCommand, CreateOrderCommand ↔ CancelOrderCommand.
• NestJS не даёт декларативных транзакций «из коробки» — используй DataSource.transaction(async (m) => { ... }) явно; никакого магического rollback нет.
• Compensation идемпотентна: перед действием проверяется статус (status === 'refunded' → return); saga может повторить compensation при retry сколько угодно раз.
• Semantic, не технический. Refund — новая запись в БД и новый вызов платёжного провайдера, не откат EntityManager-транзакции.
• Audit trail обязателен: compensation меняет статус на refunded со ссылкой на оригинал; repository.delete запрещён.
• Failure compensation — терминальный сценарий: refund упал → status = 'compensation_failed' → DLQ-топик → алерт; silent fail недопустим.
• Compensation-команды публикуются через outbox в той же DataSource.transaction, в которой обновляется статус саги.

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

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

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

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

// src/saga/order-saga.orchestrator.ts

@Injectable()
export class OrderSagaOrchestrator {
  constructor(
    private readonly dataSource: DataSource,
    private readonly outbox: OutboxWriter,
  ) {}

  async onInventoryFailed(event: InventoryFailedEvent): Promise<void> {
    await this.dataSource.transaction(async (m) => {
      await m.update(OrderSagaEntity, { sagaId: event.sagaId }, {
        status: 'compensating',
        currentStep: 'refund_payment',
      });
      await this.outbox.write(m, {
        sagaId: event.sagaId,
        topic: 'payment.commands',
        payload: new RefundPaymentCommand(event.sagaId, event.orderId, 'inventory_unavailable'),
      });
    });
  }
}

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

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

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

// src/payment/handlers/refund-payment.handler.ts

@Injectable()
export class RefundPaymentHandler {
  constructor(
    private readonly dataSource: DataSource,
    private readonly paymentRepository: PaymentRepository,
    private readonly paymentProvider: PaymentProviderPort,
    private readonly outbox: OutboxWriter,
  ) {}

  async handle(command: RefundPaymentCommand): Promise<void> {
    await this.dataSource.transaction(async (m) => {
      const payment = await m.getRepository(PaymentEntity).findOne({
        where: { orderId: command.orderId },
        lock: { mode: 'pessimistic_write' },
      });
      if (!payment) throw new PaymentNotFoundException(command.orderId);

      if (payment.status === 'refunded') {
        return;
      }
      if (payment.status !== 'charged') {
        throw new PaymentNotRefundableException(payment.id, payment.status);
      }

      const refundId = await this.paymentProvider.refund(
        payment.externalId,
        payment.amount,
        command.sagaId,
      );

      await m.update(PaymentEntity, { id: payment.id }, {
        status: 'refunded',
        refundId,
        refundReason: command.reason,
        refundedAt: new Date(),
      });

      await this.outbox.write(m, {
        sagaId: command.sagaId,
        topic: 'payment.events',
        payload: new PaymentRefundedEvent(command.sagaId, command.orderId, refundId),
      });
    });
  }
}

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

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

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

Forward:      RequestPayment  → 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()

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

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

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

// src/order/handlers/cancel-order.handler.ts

@Injectable()
export class CancelOrderHandler {
  constructor(private readonly dataSource: DataSource) {}

  async handle(command: CancelOrderCommand): Promise<void> {
    await this.dataSource.transaction(async (m) => {
      const result = await m
        .createQueryBuilder()
        .update(OrderEntity)
        .set({
          status: 'cancelled',
          cancelReason: command.reason,
          cancelledAt: new Date(),
        })
        .where('id = :id AND status NOT IN (:...excluded)', {
          id: command.orderId,
          excluded: ['cancelled', 'completed'],
        })
        .execute();

      if (result.affected === 0) {
        return;
      }

      await m.insert(OutboxEventEntity, toOutbox(
        new OrderCancelledEvent(command.orderId, command.sagaId, command.reason),
      ));
    });
  }
}

Условие status NOT IN ('cancelled', 'completed') — защита от гонки: если параллельный поток уже отменил заказ, affected === 0, и handler завершается как no-op (идемпотентность).

Saga завершает compensation

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

// src/saga/order-saga.orchestrator.ts

async onPaymentRefunded(event: PaymentRefundedEvent): Promise<void> {
  await this.dataSource.transaction(async (m) => {
    await m.update(OrderSagaEntity, { sagaId: event.sagaId }, {
      status: 'failed',
      currentStep: 'terminal',
    });
    await m
      .createQueryBuilder()
      .update(OrderEntity)
      .set({ status: 'cancelled', cancelReason: 'payment_refunded', cancelledAt: new Date() })
      .where('id = :id', { id: event.orderId })
      .execute();
  });
}

Failure compensation — DLQ

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

// src/saga/saga-consumer.service.ts

@Injectable()
export class SagaConsumerService implements OnModuleInit {
  constructor(
    private readonly kafka: Kafka,
    private readonly dataSource: DataSource,
    private readonly dlqProducer: DlqProducer,
    private readonly orchestrator: OrderSagaOrchestrator,
  ) {}

  private async handleWithDLQ(message: KafkaMessage, sagaId: string): Promise<void> {
    try {
      await this.orchestrator.dispatch(message);
    } catch (err) {
      this.logger.error('compensation failed, routing to DLQ', { sagaId, error: err });

      await this.dataSource.transaction(async (m) => {
        await m.update(OrderSagaEntity, { sagaId }, {
          status: 'compensation_failed',
          currentStep: 'terminal',
        });
      });

      await this.dlqProducer.send({
        topic: 'payment.compensation.dlq',
        messages: [{ key: sagaId, value: JSON.stringify({ sagaId, error: String(err) }) }],
      });
    }
  }
}

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

Без DLQ:

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

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

Куда дальше

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