← Volver al listado de tecnologías

Capítulo 4: Transacciones Compensatorias

Por: SiempreListo
sagacompensaciónidempotenciarollback

Capítulo 4: Transacciones Compensatorias

“Cada acción debe tener su antídoto”

Introducción

En las transacciones tradicionales de base de datos, cuando algo falla, el sistema simplemente hace un “rollback” que deshace automáticamente todos los cambios. Pero en las sagas, cada paso se confirma inmediatamente en su propia base de datos. No hay un mecanismo automático para deshacer.

Aquí es donde entran las transacciones compensatorias: operaciones de negocio diseñadas específicamente para revertir el efecto de una transacción anterior.

Qué es una Compensación

Una transacción compensatoria deshace semánticamente el efecto de una transacción anterior. “Semánticamente” significa que revierte el significado de negocio, no necesariamente los bytes exactos en la base de datos:

T1: Reservar stock → C1: Liberar stock
T2: Cobrar pago   → C2: Reembolsar pago
T3: Crear envío   → C3: Cancelar envío

Importante: No es un rollback de base de datos. Es una operación de negocio que revierte el efecto. Por ejemplo, la compensación de “enviar email de confirmación” no puede “des-enviar” el email, pero podría enviar un email de cancelación.

Diseño de Compensaciones

TypeScript

interface CompensableAction<TInput, TOutput> {
  name: string;
  execute(input: TInput): Promise<TOutput>;
  compensate(output: TOutput): Promise<void>;
  isCompensatable(output: TOutput): boolean;
}

class ReserveStockAction implements CompensableAction<ReserveInput, Reservation> {
  name = 'reserveStock';

  async execute(input: ReserveInput): Promise<Reservation> {
    const reservation = await this.db.insert(reservations).values({
      id: crypto.randomUUID(),
      orderId: input.orderId,
      productId: input.productId,
      quantity: input.quantity,
      status: 'active',
      createdAt: new Date()
    }).returning();

    await this.db.update(inventory)
      .set({ reserved: sql`reserved + ${input.quantity}` })
      .where(eq(inventory.productId, input.productId));

    return reservation[0];
  }

  async compensate(reservation: Reservation): Promise<void> {
    await this.db.update(reservations)
      .set({ status: 'cancelled' })
      .where(eq(reservations.id, reservation.id));

    await this.db.update(inventory)
      .set({ reserved: sql`reserved - ${reservation.quantity}` })
      .where(eq(inventory.productId, reservation.productId));
  }

  isCompensatable(reservation: Reservation): boolean {
    return reservation.status === 'active';
  }
}

Go

type CompensableAction[TInput, TOutput any] interface {
    Name() string
    Execute(ctx context.Context, input TInput) (TOutput, error)
    Compensate(ctx context.Context, output TOutput) error
    IsCompensatable(output TOutput) bool
}

type ReserveStockAction struct {
    db *sql.DB
}

func (a *ReserveStockAction) Name() string {
    return "reserveStock"
}

func (a *ReserveStockAction) Execute(ctx context.Context, input ReserveInput) (*Reservation, error) {
    tx, err := a.db.BeginTx(ctx, nil)
    if err != nil {
        return nil, err
    }
    defer tx.Rollback()

    reservation := &Reservation{
        ID:        uuid.New().String(),
        OrderID:   input.OrderID,
        ProductID: input.ProductID,
        Quantity:  input.Quantity,
        Status:    "active",
        CreatedAt: time.Now(),
    }

    _, err = tx.ExecContext(ctx, `
        INSERT INTO reservations (id, order_id, product_id, quantity, status, created_at)
        VALUES ($1, $2, $3, $4, $5, $6)`,
        reservation.ID, reservation.OrderID, reservation.ProductID,
        reservation.Quantity, reservation.Status, reservation.CreatedAt)
    if err != nil {
        return nil, err
    }

    _, err = tx.ExecContext(ctx, `
        UPDATE inventory SET reserved = reserved + $1 WHERE product_id = $2`,
        input.Quantity, input.ProductID)
    if err != nil {
        return nil, err
    }

    return reservation, tx.Commit()
}

func (a *ReserveStockAction) Compensate(ctx context.Context, r *Reservation) error {
    tx, err := a.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    _, err = tx.ExecContext(ctx, `UPDATE reservations SET status = 'cancelled' WHERE id = $1`, r.ID)
    if err != nil {
        return err
    }

    _, err = tx.ExecContext(ctx, `UPDATE inventory SET reserved = reserved - $1 WHERE product_id = $2`,
        r.Quantity, r.ProductID)
    if err != nil {
        return err
    }

    return tx.Commit()
}

func (a *ReserveStockAction) IsCompensatable(r *Reservation) bool {
    return r.Status == "active"
}

Python

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Generic, TypeVar
import uuid
from datetime import datetime

TInput = TypeVar('TInput')
TOutput = TypeVar('TOutput')

class CompensableAction(ABC, Generic[TInput, TOutput]):
    @property
    @abstractmethod
    def name(self) -> str:
        pass

    @abstractmethod
    async def execute(self, input_data: TInput) -> TOutput:
        pass

    @abstractmethod
    async def compensate(self, output: TOutput) -> None:
        pass

    @abstractmethod
    def is_compensatable(self, output: TOutput) -> bool:
        pass


@dataclass
class Reservation:
    id: str
    order_id: str
    product_id: str
    quantity: int
    status: str
    created_at: datetime


class ReserveStockAction(CompensableAction[dict, Reservation]):
    def __init__(self, db):
        self.db = db

    @property
    def name(self) -> str:
        return "reserveStock"

    async def execute(self, input_data: dict) -> Reservation:
        async with self.db.transaction():
            reservation = Reservation(
                id=str(uuid.uuid4()),
                order_id=input_data['order_id'],
                product_id=input_data['product_id'],
                quantity=input_data['quantity'],
                status='active',
                created_at=datetime.now()
            )

            await self.db.execute("""
                INSERT INTO reservations (id, order_id, product_id, quantity, status, created_at)
                VALUES ($1, $2, $3, $4, $5, $6)
            """, reservation.id, reservation.order_id, reservation.product_id,
                reservation.quantity, reservation.status, reservation.created_at)

            await self.db.execute("""
                UPDATE inventory SET reserved = reserved + $1 WHERE product_id = $2
            """, input_data['quantity'], input_data['product_id'])

            return reservation

    async def compensate(self, reservation: Reservation) -> None:
        async with self.db.transaction():
            await self.db.execute(
                "UPDATE reservations SET status = 'cancelled' WHERE id = $1",
                reservation.id
            )
            await self.db.execute(
                "UPDATE inventory SET reserved = reserved - $1 WHERE product_id = $2",
                reservation.quantity, reservation.product_id
            )

    def is_compensatable(self, reservation: Reservation) -> bool:
        return reservation.status == 'active'

Idempotencia

La idempotencia es una propiedad fundamental en sistemas distribuidos. Una operación es idempotente si ejecutarla una vez produce el mismo resultado que ejecutarla múltiples veces.

¿Por qué importa? En un sistema distribuido, los mensajes pueden llegar duplicados, las conexiones pueden cortarse a mitad de la operación, y los reintentos son comunes. Si una compensación no es idempotente, ejecutarla dos veces podría causar problemas (como liberar el stock dos veces).

Las compensaciones deben ser idempotentes: ejecutarlas múltiples veces produce el mismo resultado.

Estrategia con Idempotency Key

Una idempotency key (clave de idempotencia) es un identificador único que asociamos con cada operación. Antes de ejecutar, verificamos si ya procesamos esa clave. Si ya existe, devolvemos el resultado anterior sin ejecutar de nuevo.

class IdempotentCompensation {
  async compensate(compensationId: string, action: () => Promise<void>): Promise<void> {
    // Verificar si ya se ejecutó
    const existing = await this.db.query.compensations.findFirst({
      where: eq(compensations.id, compensationId)
    });

    if (existing?.status === 'completed') {
      console.log(`Compensation ${compensationId} already executed`);
      return;
    }

    // Marcar como en progreso
    await this.db.insert(compensations).values({
      id: compensationId,
      status: 'in_progress',
      startedAt: new Date()
    }).onConflictDoUpdate({
      target: compensations.id,
      set: { status: 'in_progress', startedAt: new Date() }
    });

    try {
      await action();

      await this.db.update(compensations)
        .set({ status: 'completed', completedAt: new Date() })
        .where(eq(compensations.id, compensationId));
    } catch (error) {
      await this.db.update(compensations)
        .set({ status: 'failed', error: String(error) })
        .where(eq(compensations.id, compensationId));
      throw error;
    }
  }
}

Go - Idempotencia con Mutex Distribuido

Un mutex (mutual exclusion) es un mecanismo de sincronización que garantiza que solo un proceso pueda ejecutar una sección de código a la vez. Un mutex distribuido extiende este concepto a múltiples servidores usando un sistema externo como Redis.

type IdempotentCompensator struct {
    db    *sql.DB
    redis *redis.Client
}

func (ic *IdempotentCompensator) Compensate(ctx context.Context, key string, action func() error) error {
    // Obtener lock distribuido
    lock, err := ic.redis.SetNX(ctx, "lock:"+key, "1", 30*time.Second).Result()
    if err != nil {
        return err
    }
    if !lock {
        return fmt.Errorf("compensation %s already in progress", key)
    }
    defer ic.redis.Del(ctx, "lock:"+key)

    // Verificar estado
    var status string
    err = ic.db.QueryRowContext(ctx,
        "SELECT status FROM compensations WHERE id = $1", key).Scan(&status)

    if err == nil && status == "completed" {
        return nil // Ya ejecutada
    }

    // Ejecutar compensación
    if err := action(); err != nil {
        ic.db.ExecContext(ctx,
            "INSERT INTO compensations (id, status) VALUES ($1, 'failed') ON CONFLICT (id) DO UPDATE SET status = 'failed'",
            key)
        return err
    }

    _, err = ic.db.ExecContext(ctx,
        "INSERT INTO compensations (id, status) VALUES ($1, 'completed') ON CONFLICT (id) DO UPDATE SET status = 'completed'",
        key)
    return err
}

Orden de Compensación

Las compensaciones se ejecutan en orden inverso (LIFO: Last In, First Out). Esto tiene sentido intuitivo: si construiste algo paso a paso, para desmontarlo debes empezar por lo último que hiciste.

Por ejemplo, si primero creaste un pedido, luego reservaste inventario, y finalmente procesaste el pago, la compensación primero revierte el pago (reembolso), luego libera el inventario, y finalmente cancela el pedido.

sequenceDiagram
    participant S as Saga
    participant T1 as Order
    participant T2 as Inventory
    participant T3 as Payment

    Note over S,T3: Ejecución normal
    S->>T1: T1: Create Order
    T1-->>S: OK
    S->>T2: T2: Reserve Stock
    T2-->>S: OK
    S->>T3: T3: Process Payment
    T3-->>S: FAIL

    Note over S,T3: Compensación (orden inverso)
    S->>T2: C2: Release Stock
    T2-->>S: OK
    S->>T1: C1: Cancel Order
    T1-->>S: OK

Implementación del Rollback

class SagaExecutor {
  async execute<T>(steps: SagaStep<T>[], context: T): Promise<void> {
    const completedSteps: Array<{ step: SagaStep<T>; result: unknown }> = [];

    for (const step of steps) {
      try {
        const result = await step.execute(context);
        completedSteps.push({ step, result });
      } catch (error) {
        await this.rollback(completedSteps, context);
        throw new SagaFailedError(step.name, error);
      }
    }
  }

  private async rollback<T>(
    completedSteps: Array<{ step: SagaStep<T>; result: unknown }>,
    context: T
  ): Promise<void> {
    const errors: Error[] = [];

    // Compensar en orden inverso
    for (const { step, result } of completedSteps.reverse()) {
      try {
        await step.compensate(context, result);
      } catch (error) {
        errors.push(new CompensationError(step.name, error));
        // Continuar con las demás compensaciones
      }
    }

    if (errors.length > 0) {
      throw new PartialCompensationError(errors);
    }
  }
}

Compensaciones Parciales

A veces las compensaciones también pueden fallar. El servicio de pagos podría estar caído cuando intentamos reembolsar. ¿Qué hacemos?

La estrategia más común es registrar la compensación fallida para reintentarla posteriormente. Un proceso en segundo plano (worker) revisa periódicamente las compensaciones pendientes y las reintenta con backoff exponencial (esperas cada vez más largas entre reintentos).

Cuando una compensación falla, se debe registrar para reintento:

interface FailedCompensation {
  sagaId: string;
  stepName: string;
  payload: unknown;
  attempts: number;
  lastError: string;
  nextRetryAt: Date;
}

class CompensationRetryWorker {
  async processFailedCompensations(): Promise<void> {
    const failed = await this.db.query.failedCompensations.findMany({
      where: and(
        lt(failedCompensations.nextRetryAt, new Date()),
        lt(failedCompensations.attempts, 5)
      )
    });

    for (const compensation of failed) {
      try {
        await this.executeCompensation(compensation);
        await this.db.delete(failedCompensations)
          .where(eq(failedCompensations.id, compensation.id));
      } catch (error) {
        await this.db.update(failedCompensations)
          .set({
            attempts: compensation.attempts + 1,
            lastError: String(error),
            nextRetryAt: this.calculateNextRetry(compensation.attempts)
          })
          .where(eq(failedCompensations.id, compensation.id));
      }
    }
  }

  private calculateNextRetry(attempts: number): Date {
    // Backoff exponencial: 1min, 2min, 4min, 8min, 16min
    // Cada reintento espera el doble que el anterior
    const delayMs = Math.pow(2, attempts) * 60 * 1000;
    return new Date(Date.now() + delayMs);
  }
}

Tipos de Compensación

TipoDescripciónEjemplo
ReversiónDeshace completamenteReembolso de pago
AnulaciónMarca como canceladoCancelar reserva
ContrarrestaciónOperación opuestaCrédito vs débito
No-opSin acción necesariaNotificación enviada

Resumen

Glosario

Transacción Compensatoria

Definición: Operación de negocio diseñada para revertir semánticamente el efecto de una transacción anterior que ya fue confirmada.

Por qué es importante: En sagas, cada paso se confirma inmediatamente. No hay rollback automático. Las compensaciones son el mecanismo para deshacer cambios cuando pasos posteriores fallan.

Ejemplo práctico: Si “reservar inventario” reservó 5 unidades del producto A, la compensación “liberar inventario” devuelve esas 5 unidades al stock disponible.

Idempotencia

Definición: Propiedad de una operación donde ejecutarla una o múltiples veces produce el mismo resultado final.

Por qué es importante: En sistemas distribuidos, los mensajes pueden duplicarse y las operaciones reintentarse. Sin idempotencia, procesar el mismo mensaje dos veces podría causar inconsistencias (como cobrar dos veces).

Ejemplo práctico: “Liberar reserva R123” es idempotente si la segunda llamada detecta que R123 ya fue liberada y simplemente retorna éxito sin hacer nada. No es idempotente si suma stock de nuevo.

Idempotency Key

Definición: Identificador único asociado con una operación que permite detectar si esa operación ya fue procesada anteriormente.

Por qué es importante: Es la técnica más común para implementar idempotencia. Antes de ejecutar, verificamos si la clave existe. Si existe, devolvemos el resultado guardado sin ejecutar de nuevo.

Ejemplo práctico: Al procesar un pago, usamos order_${orderId} como idempotency key. Si llega una segunda solicitud con el mismo orderId, reconocemos que ya procesamos este pago y devolvemos el resultado anterior.

Rollback

Definición: En bases de datos tradicionales, es la operación que deshace todos los cambios de una transacción que no se completó exitosamente.

Por qué es importante: El rollback es automático en transacciones ACID. En sagas no existe este mecanismo automático; debemos implementar compensaciones explícitas.

Ejemplo práctico: En SQL, si ejecutas BEGIN; INSERT...; UPDATE...; ROLLBACK;, tanto el INSERT como el UPDATE se deshacen. En una saga, tendrías que llamar explícitamente a las compensaciones de cada operación.

Semántica de Negocio

Definición: El significado o intención de una operación desde la perspectiva del negocio, no de la implementación técnica.

Por qué es importante: Las compensaciones revierten la semántica, no necesariamente los datos exactos. Por ejemplo, un reembolso no “des-procesa” el pago original; crea una nueva transacción que revierte el efecto.

Ejemplo práctico: La semántica de “cobrar $100” es “el cliente tiene $100 menos”. La compensación semántica es “el cliente tiene $100 más” (reembolso), no “borrar el registro del cobro”.

Mutex (Mutual Exclusion)

Definición: Mecanismo de sincronización que garantiza que solo un proceso o hilo pueda acceder a un recurso o ejecutar una sección de código en un momento dado.

Por qué es importante: Previene condiciones de carrera donde dos procesos intentan modificar el mismo recurso simultáneamente, causando inconsistencias.

Ejemplo práctico: Si dos instancias del servicio intentan compensar la misma reserva simultáneamente, un mutex distribuido (usando Redis) garantiza que solo una lo haga.

Lock Distribuido

Definición: Mutex implementado para funcionar en múltiples servidores, típicamente usando un sistema de almacenamiento compartido como Redis o ZooKeeper.

Por qué es importante: En arquitecturas de microservicios con múltiples instancias, un mutex local no es suficiente. El lock distribuido coordina entre todas las instancias.

Ejemplo práctico: Antes de ejecutar una compensación, obtenemos un lock con clave lock:compensation:R123 en Redis. Si otra instancia ya tiene el lock, esperamos o abortamos.

Backoff Exponencial

Definición: Estrategia de reintento donde el tiempo de espera entre intentos se duplica (o multiplica) con cada fallo consecutivo.

Por qué es importante: Evita sobrecargar un servicio que está fallando. Si el servicio está caído, bombardearlo con reintentos inmediatos solo empeora las cosas.

Ejemplo práctico: Primer reintento después de 1 minuto, segundo después de 2 minutos, tercero después de 4 minutos, cuarto después de 8 minutos, hasta un máximo (por ejemplo, 16 minutos).

Orden Inverso (LIFO)

Definición: Las compensaciones se ejecutan en orden inverso al de las transacciones originales. El último paso ejecutado es el primero en compensarse.

Por qué es importante: Mantiene la coherencia lógica. Los pasos posteriores pueden depender de los anteriores, así que debemos deshacer primero los que dependen antes de deshacer aquellos de los que dependen.

Ejemplo práctico: Si ejecutamos T1 (crear pedido) -> T2 (reservar stock) -> T3 (procesar pago) y T3 falla, compensamos en orden: C2 (liberar stock) -> C1 (cancelar pedido). No compensamos T3 porque nunca se completó.

Compensación No-op

Definición: Compensación que no requiere ninguna acción porque el efecto original no puede o no necesita revertirse.

Por qué es importante: No todas las operaciones tienen una compensación significativa. Reconocer esto evita escribir código innecesario y simplifica el diseño.

Ejemplo práctico: Si enviamos una notificación por email, no podemos “des-enviarla”. La compensación podría ser no-op (no hacer nada) o enviar otro email explicando la cancelación.

Worker de Compensación

Definición: Proceso en segundo plano que revisa periódicamente las compensaciones fallidas y las reintenta.

Por qué es importante: Garantiza que las compensaciones eventualmente se completen, incluso si fallan inicialmente por problemas temporales de red o servicios.

Ejemplo práctico: Cada 30 segundos, el worker consulta la tabla failed_compensations, selecciona las que tienen next_retry_at < NOW() y menos de 5 intentos, y las ejecuta.

← Capítulo 3: Orquestación vs Coreografía | Capítulo 5: Diseño Resiliente →

← Volver al listado de tecnologías