Patrón de Buzón de Salida Transaccional en Go con PostgreSQL

Dos escrituras que deberían tener éxito juntas eventualmente fallarán por separado. Tu servicio de pedidos guarda el pedido en la base de datos y luego publica un evento order.created en un intermediario de mensajes.

Estas dos operaciones se ejecutan una tras otra.

Entre ambas, las cosas salen mal: el intermediario está caído, la red pierde conexión, el proceso se reinicia o el contenedor es expulsado. La escritura en la base de datos fue exitosa. La publicación no lo fue. El servicio aguas abajo que necesita saber sobre el nuevo pedido nunca se entera. Nadie se dio cuenta hasta que un cliente llamó.

Este es el problema de la escritura dual, y es una de las fuentes más comunes de pérdida silenciosa de datos en sistemas distribuidos. El patrón de buzón de salida transaccional es la solución estándar.

El problema de la escritura dual

El modo de fallo es fácil de comprender una vez que lo ves:

BEGIN;
  INSERT INTO orders ...   -- tiene éxito
COMMIT;

PUBLISH order.created ...  -- falla, se bloquea o nunca se alcanza

La base de datos y el intermediario de mensajes no comparten un límite de transacción. No hay un retroceso (rollback) que cubra a ambos. Cada servicio que hace guardar -> publicar en secuencia tiene esta brecha. El patrón aparece en muchas formas:

• db.Save(order) seguido de events.Publish(OrderCreated{...})
• Manejador HTTP que confirma una transacción y luego llama a un webhook externo
• Trabajador que procesa un registro de una cola y escribe resultados en otra

El resultado en todos los casos es el mismo: un lado tiene éxito mientras el otro falla, y el sistema termina en un estado que es invisible para la monitorización porque ambas operaciones individuales devolvieron éxito en algún momento.

Un bucle de reintento no soluciona esto. Reintentar la publicación después del compromiso de la base de datos solo funciona si el reintento en sí es confiable, lo cual requiere exactamente la garantía de durabilidad que no tienes.

Qué hace el patrón de buzón de salida transaccional

El patrón de buzón de salida elimina la brecha eliminando por completo la publicación directa. En lugar de llamar al intermediario desde dentro de tu lógica de negocio, escribes un registro de evento en una tabla outbox en la misma transacción de base de datos que los datos de negocio. Un proceso de fondo separado, el retransmisor (relay), lee de la tabla de buzón de salida y publica en el intermediario.

BEGIN;
  INSERT INTO orders ...         -- datos de negocio
  INSERT INTO outbox_events ...  -- registro de evento
COMMIT;

-- Proceso de retransmisión (por separado):
SELECT ... FROM outbox_events FOR UPDATE SKIP LOCKED;
PUBLISH order.created ...
UPDATE outbox_events SET processed_at = NOW() WHERE id = $1;

Ambas escrituras tienen éxito o ambas fallan. La garantía de transacción que ya tienes de PostgreSQL ahora también cubre el registro de evento. El retransmisor puede reintentar la publicación tantas veces como sea necesario porque el evento reside en almacenamiento durable. Si el retransmisor falla en medio del proceso, se reinicia y reintenta. El peor resultado es que el evento se publique más de una vez, lo cual se maneja haciendo que los consumidores sean idempotentes (ver Idempotencia en Sistemas Distribuidos).

Esquema de PostgreSQL para la tabla de buzón de salida

El esquema es intencionalmente simple:

CREATE TABLE outbox_events (
    id             UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        JSONB        NOT NULL,
    attempts       INT          NOT NULL DEFAULT 0,
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    processed_at   TIMESTAMPTZ
);

-- Índice parcial: solo indexa filas no procesadas, se mantiene pequeño a medida que las filas se marcan como completadas
CREATE INDEX idx_outbox_unprocessed
    ON outbox_events (created_at)
    WHERE processed_at IS NULL;

El índice parcial en created_at WHERE processed_at IS NULL es importante. Sin él, el índice crece con cada evento escrito y la consulta de sondeo del retransmisor se vuelve más lenta con el tiempo. Con él, el índice cubre solo las filas pendientes, que en estado estable son un conjunto pequeño y acotado, independientemente de cuántos eventos se hayan publicado.

Elecciones clave de campos:

• aggregate_type y aggregate_id describen a qué entidad pertenece el evento. Útil para garantías de ordenamiento y enrutamiento.
• event_type es el nombre del evento que esperan tus consumidores.
• payload JSONB almacena el cuerpo del evento. Usa JSONB en lugar de TEXT para poder consultarlo si es necesario.
• attempts rastrea cuántas veces el retransmisor ha intentado publicar esta fila. Se usa para límites de reintento y manejo de mensajes muertos.
• processed_at es NULL para filas pendientes y se establece cuando el retransmisor publica con éxito.

Escribir datos de negocio y evento de buzón de salida en una sola transacción

La lógica de negocio escribe ambos registros dentro de una sola llamada BeginTx / Commit. No hay llamada de publicación aquí, solo escrituras en la base de datos.

type OrderService struct {
    db *sql.DB
}

func (s *OrderService) CreateOrder(ctx context.Context, order Order) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback()

    if _, err := tx.ExecContext(ctx, `
        INSERT INTO orders (id, customer_id, total, created_at)
        VALUES ($1, $2, $3, NOW())
    `, order.ID, order.CustomerID, order.Total); err != nil {
        return fmt.Errorf("insert order: %w", err)
    }

    payload, err := json.Marshal(map[string]any{
        "order_id":    order.ID,
        "customer_id": order.CustomerID,
        "total":       order.Total,
    })
    if err != nil {
        return fmt.Errorf("marshal payload: %w", err)
    }

    if _, err := tx.ExecContext(ctx, `
        INSERT INTO outbox_events
            (aggregate_type, aggregate_id, event_type, payload)
        VALUES ($1, $2, $3, $4)
    `, "order", order.ID, "order.created", payload); err != nil {
        return fmt.Errorf("insert outbox event: %w", err)
    }

    return tx.Commit()
}

Si tx.Commit() falla, ni la fila del pedido ni la fila del buzón de salida se persisten. Si tiene éxito, se garantiza que ambas están en la base de datos. El retransmisor puede publicar el evento en cualquier momento después de eso: inmediatamente, en un segundo, o después de que el retransmisor se reinicie tras un fallo.

Este es el único cambio de código requerido en tu capa de negocio. El resto del patrón reside en el retransmisor.

Implementación del retransmisor en Go

El retransmisor es un trabajador de fondo que sondea la tabla de buzón de salida en un temporizador. Obtiene un lote de filas no procesadas, publica cada una y las marca como completadas. Manténlo en el mismo binario que tu aplicación o ejecútalo como un proceso separado; cualquiera funciona, pero el mismo binario es más simple de operar.

type OutboxRelay struct {
    db          *sql.DB
    publisher   Publisher
    logger      *slog.Logger
    batchSize   int
    pollInterval time.Duration
    maxAttempts  int
}

func (r *OutboxRelay) Run(ctx context.Context) error {
    ticker := time.NewTicker(r.pollInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-ticker.C:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox relay batch failed", "err", err)
            }
        }
    }
}

El retransmisor respeta la cancelación del contexto, lo que facilita su integración con el apagado ordenado. Para un tratamiento detallado del ciclo de vida del contexto y los patrones de cancelación, ver Go context.Context Done Right.

FOR UPDATE SKIP LOCKED: el patrón de trabajador concurrente

La función processBatch usa FOR UPDATE SKIP LOCKED para manejar de forma segura los trabajadores de retransmisión concurrentes:

func (r *OutboxRelay) processBatch(ctx context.Context) error {
    tx, err := r.db.BeginTx(ctx, nil)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback()

    rows, err := tx.QueryContext(ctx, `
        SELECT id, aggregate_type, aggregate_id, event_type, payload
        FROM outbox_events
        WHERE processed_at IS NULL
          AND attempts < $1
        ORDER BY created_at
        LIMIT $2
        FOR UPDATE SKIP LOCKED
    `, r.maxAttempts, r.batchSize)
    if err != nil {
        return fmt.Errorf("query outbox: %w", err)
    }
    defer rows.Close()

    type row struct {
        id            string
        aggregateType string
        aggregateID   string
        eventType     string
        payload       json.RawMessage
    }

    var batch []row
    for rows.Next() {
        var e row
        if err := rows.Scan(
            &e.id, &e.aggregateType, &e.aggregateID, &e.eventType, &e.payload,
        ); err != nil {
            return fmt.Errorf("scan row: %w", err)
        }
        batch = append(batch, e)
    }
    if err := rows.Err(); err != nil {
        return err
    }

    for _, e := range batch {
        if err := r.publisher.Publish(ctx, e.eventType, e.aggregateID, e.payload); err != nil {
            r.logger.Error("publish failed", "event_id", e.id, "err", err)
            if _, err := tx.ExecContext(ctx,
                `UPDATE outbox_events SET attempts = attempts + 1 WHERE id = $1`, e.id,
            ); err != nil {
                r.logger.Error("increment attempts failed", "event_id", e.id, "err", err)
            }
            continue
        }

        if _, err := tx.ExecContext(ctx,
            `UPDATE outbox_events SET processed_at = NOW() WHERE id = $1`, e.id,
        ); err != nil {
            return fmt.Errorf("mark processed: %w", err)
        }
    }

    return tx.Commit()
}

FOR UPDATE SKIP LOCKED hace dos cosas. Primero, FOR UPDATE bloquea las filas seleccionadas durante la duración de la transacción, impidiendo que cualquier otra transacción las seleccione. Segundo, SKIP LOCKED significa que si una fila ya está bloqueada por otra transacción, la consulta la omite en lugar de esperar. El resultado es que múltiples trabajadores de retransmisión pueden ejecutarse en paralelo y cada uno seleccionará un subconjunto de filas que no se superpongan.

Sin SKIP LOCKED, un segundo trabajador bloquearía hasta que la primera transacción se confirmara antes de ver las mismas filas, momento en el cual ya estarían marcadas como completadas. Con SKIP LOCKED, el segundo trabajador selecciona inmediatamente filas diferentes en lugar de esperar, lo que te ofrece escalabilidad horizontal segura.

Observa la separación de escaneo y publicación en el código anterior: todas las filas se escanean en una porción (slice) antes de que comience el bucle de publicación. Esto evita mantener un cursor *sql.Rows abierto durante las llamadas de red al intermediario, lo cual mantendría la transacción abierta más tiempo del necesario.

Idempotencia y deduplicación

El retransmisor publica al menos una vez. Si publica un evento y luego falla antes de confirmar la actualización de processed_at, publicará el mismo evento nuevamente al reiniciar. Esto es inevitable: la entrega exactamente una vez a través de una base de datos y un intermediario de mensajes sin un coordinador de transacciones distribuidas requiere este compromiso.

Los consumidores deben ser idempotentes. El enfoque más simple es rastrear los IDs de eventos procesados en una tabla processed_events:

CREATE TABLE processed_events (
    event_id   UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

func (h *OrderHandler) HandleOrderCreated(ctx context.Context, eventID string, payload []byte) error {
    // Deduplicar usando el ID del evento como clave natural
    _, err := h.db.ExecContext(ctx, `
        INSERT INTO processed_events (event_id) VALUES ($1)
        ON CONFLICT (event_id) DO NOTHING
    `, eventID)
    if err != nil {
        return fmt.Errorf("dedup check: %w", err)
    }

    // Verificar si la inserción realmente ocurrió (1 fila) o fue una operación nula (0 filas)
    // Un enfoque más simple: usar RETURNING o verificar filas afectadas
    // Si 0 filas afectadas, es un duplicado: omítelo
    ...
}

En la práctica, muchos equipos confían en los propios encabezados de deduplicación del intermediario (como el campo key de Kafka para temas compactados por registro, o el encabezado message-id de RabbitMQ) y tratan la deduplicación a nivel de base de datos como una solución de respaldo. Ambas son capas válidas para aplicar.

Incluye el id del evento de buzón de salida (un UUID) en el mensaje publicado como clave de deduplicación. Los consumidores pueden usarlo independientemente del mecanismo de deduplicación que prefieran.

Política de reintento y mensajes tóxicos

La columna attempts impulsa la política de reintento. El retransmisor omite filas donde attempts >= maxAttempts y trata esas filas como mensajes muertos. Un proceso separado o una alerta de operador se encarga de ellas.

Una vista simple de mensajes muertos:

CREATE VIEW outbox_dead_letters AS
SELECT *
FROM outbox_events
WHERE attempts >= 5
  AND processed_at IS NULL
ORDER BY created_at;

Una buena política de reintento en producción:

• Establece maxAttempts en 5-10 dependiendo de qué tan costosos sean los reintentos.
• Considera el retroceso exponencial: incluye una columna retry_after y omite filas donde retry_after > NOW().
• Alerta cuando COUNT(*) FROM outbox_dead_letters exceda un umbral.
• Proporciona una ruta de reintento manual: un punto final de administrador o script que restablezca attempts = 0 y retry_after = NULL para filas específicas.

Los mensajes tóxicos, filas que fallan consistentemente debido a un error en el consumidor o una incompatibilidad de esquema, no deben bloquear los mensajes saludables. Dado que el retransmisor procesa un lote por tick y marca los fallos con un incremento de intentos en lugar de eliminarlos de la cola, las filas saludables proceden normalmente mientras las tóxicas acumulan intentos hasta alcanzar el umbral de mensajes muertos.

Ordenamiento de eventos y particionamiento

La consulta de sondeo ordena por created_at, lo que proporciona un ordenamiento primero en entrar, primero en salir dentro de un lote. Para la mayoría de los casos de uso, eso es suficiente. Cuando el ordenamiento estricto por entidad importa, por ejemplo, asegurando que order.updated nunca se publique antes que order.created para el mismo pedido, necesitas ordenamiento por agregado.

Agrega aggregate_id a la cláusula ORDER BY y úsalo como clave de mensaje al publicar en un tema particionado como Apache Kafka. Kafka enruta todos los mensajes con la misma clave a la misma partición, y las particiones se consumen en orden. Esto te ofrece garantías de ordenamiento por agregado sin ordenamiento global, lo cual requeriría una única instancia de retransmisión.

ORDER BY aggregate_id, created_at

Para intermediarios que no admiten ordenamiento particionado (como colas AMQP básicas), una instancia única de retransmisión o verificaciones de ordenamiento a nivel de aplicación en el consumidor son las alternativas prácticas.

Reducir la latencia de sondeo con LISTEN/NOTIFY

Un intervalo de sondeo de un segundo significa una latencia de evento promedio de 500 milisegundos. Para la mayoría de las cargas de trabajo, eso está bien. Para casos donde necesitas latencia cercana a cero, el mecanismo LISTEN/NOTIFY de PostgreSQL permite que el retransmisor se despierte inmediatamente cuando se inserta una nueva fila de buzón de salida.

Agrega un disparador a la tabla de buzón de salida:

CREATE OR REPLACE FUNCTION notify_outbox_insert() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('outbox_event', NEW.id::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER outbox_insert_notify
AFTER INSERT ON outbox_events
FOR EACH ROW EXECUTE FUNCTION notify_outbox_insert();

En el retransmisor, escucha en el canal y despierta en las notificaciones mientras aún cae en el sondeo periódico como respaldo:

func (r *OutboxRelay) Run(ctx context.Context) error {
    listener := pq.NewListener(r.dsn, 10*time.Second, time.Minute, nil)
    defer listener.Close()

    if err := listener.Listen("outbox_event"); err != nil {
        return fmt.Errorf("listen: %w", err)
    }

    ticker := time.NewTicker(5 * time.Second) // sondeo de respaldo
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-listener.Notify:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox batch failed (notify)", "err", err)
            }
        case <-ticker.C:
            if err := r.processBatch(ctx); err != nil {
                r.logger.Error("outbox batch failed (poll)", "err", err)
            }
        }
    }
}

El temporizador de respaldo maneja cualquier notificación perdida durante un reinicio del retransmisor o un problema de red. Mantén el intervalo de respaldo en unos pocos segundos en lugar de milisegundos; su trabajo es la recuperación, no la baja latencia.

Observabilidad: métricas, registros y alertas

El buzón de salida es infraestructura. Trátalo como infraestructura e instrumentalo en consecuencia.

Métricas clave:

var (
    outboxPublished = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "outbox_events_published_total",
        Help: "Total outbox events successfully published.",
    })
    outboxFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
        Name: "outbox_events_failed_total",
        Help: "Total outbox publish failures by event type.",
    }, []string{"event_type"})
    outboxPending = prometheus.NewGauge(prometheus.GaugeOpts{
        Name: "outbox_events_pending",
        Help: "Current number of unprocessed outbox events.",
    })
    outboxBatchDuration = prometheus.NewHistogram(prometheus.HistogramOpts{
        Name:    "outbox_batch_duration_seconds",
        Help:    "Duration of each outbox processing batch.",
        Buckets: prometheus.DefBuckets,
    })
)

Actualización del medidor: ejecuta una consulta periódica para mantener outbox_events_pending preciso:

SELECT COUNT(*) FROM outbox_events WHERE processed_at IS NULL;

Umbrales de alerta a considerar:

• outbox_events_pending > 1000 durante más de dos minutos: el retransmisor se está quedando atrás o está atascado.
• outbox_events_pending creciendo monótonamente: el intermediario está caído o el retransmisor ha fallado.
• Conteo de mensajes muertos no nulo: un error de esquema o del consumidor necesita investigación.
• outbox_batch_duration_seconds p95 > 5s: la base de datos es lenta o el tamaño del lote es demasiado grande.

Campos de registro estructurados: incluye event_id, event_type, aggregate_id y attempt en cada línea de registro del retransmisor. Estos campos te permiten correlacionar una publicación fallida con la fila específica del buzón de salida y la traza del consumidor aguas abajo.

Buzón de salida vs. cola directa vs. saga

El patrón de buzón de salida no es la herramienta adecuada para cada problema de coordinación. Aquí está la comparación:

El patrón de buzón de salida garantiza que un solo servicio emita de manera confiable eventos que reflejen sus propios cambios de estado. No coordina cambios de estado a través de múltiples servicios; eso es para lo que está diseñado el patrón Saga. La elección del intermediario, ya sea RabbitMQ, SQS, o Kafka, es independiente del patrón de buzón de salida en sí; el retransmisor publica en el intermediario que use tu sistema.

Si estás construyendo un saga, el patrón de buzón de salida sigue siendo útil: cada participante en el saga escribe su cambio de estado local y su evento de saga en una sola transacción usando el buzón de salida, y luego el orquestador o coreografía del saga lee esos eventos de manera confiable.

CDC basado en WAL como retransmisor alternativo

En lugar de sondeo, puedes seguir el Registro de Antemano de Escritura (WAL) de PostgreSQL y leer las inserciones del buzón de salida directamente desde el flujo de replicación. Herramientas como Debezium hacen esto. Las ventajas son menor latencia y sin presión de bloqueo en la tabla de buzón de salida. Las desventajas son la complejidad operativa, un slot de replicación de PostgreSQL dedicado y un servicio externo para ejecutar y monitorear.

Para la mayoría de los equipos, el retransmisor de sondeo descrito anteriormente es el punto de partida correcto. El seguimiento de WAL tiene sentido cuando tienes altas tasas de inserción en el buzón de salida (decenas de miles por segundo), necesitas latencia de eventos inferior a 100 ms o ya estás ejecutando Debezium para otras necesidades de captura de cambios.

Integración con sqlc

Si usas sqlc para código de base de datos de Go a prueba de tipos, las consultas del buzón de salida encajan naturalmente:

-- name: InsertOutboxEvent :exec
INSERT INTO outbox_events (aggregate_type, aggregate_id, event_type, payload)
VALUES (@aggregate_type, @aggregate_id, @event_type, @payload);

-- name: FetchOutboxBatch :many
SELECT id, aggregate_type, aggregate_id, event_type, payload
FROM outbox_events
WHERE processed_at IS NULL
  AND attempts < @max_attempts
ORDER BY created_at
LIMIT @batch_size
FOR UPDATE SKIP LOCKED;

-- name: MarkOutboxProcessed :exec
UPDATE outbox_events SET processed_at = NOW() WHERE id = @id;

-- name: IncrementOutboxAttempts :exec
UPDATE outbox_events SET attempts = attempts + 1 WHERE id = @id;

-- name: OutboxPendingCount :one
SELECT COUNT(*) FROM outbox_events WHERE processed_at IS NULL;

sqlc genera funciones a prueba de tipos para cada consulta, lo que evita errores de interpolación de cadenas y mantiene la lógica de consulta del buzón de salida junto con el resto de tu capa de acceso a la base de datos.

Lista de verificación para producción

Usa esto antes de implementar un buzón de salida:

Base de datos

• La tabla de buzón de salida tiene el índice parcial en created_at WHERE processed_at IS NULL
• La columna attempts está presente con un valor predeterminado de 0
• Vista o consulta de mensajes muertos definida
• Las filas procesadas antiguas se archivan o eliminan periódicamente (un trabajo de limpieza nocturna es suficiente)

Retransmisor

• FOR UPDATE SKIP LOCKED usado en la consulta de sondeo
• El retransmisor se ejecuta dentro de una transacción (comenzar antes de la consulta, confirmar después de todas las actualizaciones)
• El tamaño del lote está acotado (50-200 filas es típico)
• El retransmisor respeta la cancelación del contexto para un apagado ordenado
• Las publicaciones fallidas incrementan attempts en lugar de causar que el lote se aborte

Idempotencia

• El mensaje publicado incluye el id del buzón de salida como clave de deduplicación
• Los consumidores son idempotentes o el intermediario proporciona deduplicación
• Ver Idempotencia en Sistemas Distribuidos para patrones de deduplicación

Observabilidad

• El medidor outbox_events_pending se monitorea y alerta
• El conteo de mensajes muertos genera alertas
• La duración del lote del retransmisor se rastrea
• Los registros estructurados incluyen event_id, event_type y aggregate_id

Operaciones

• Existe una ruta de reintento manual para filas de mensajes muertos
• El comportamiento de reinicio del retransmisor ha sido probado (¿republica correctamente?)
• El comportamiento durante una caída del intermediario ha sido probado (¿el buzón de salida crece y se drena correctamente?)

Pensamientos finales

El problema de la escritura dual es fácil de descartar como un caso extremo hasta que causa un incidente. El patrón de buzón de salida transaccional lo resuelve con herramientas que ya tienes: una transacción de PostgreSQL, una goroutine de fondo y una tabla extra. El retransmisor es simple de construir, simple de operar y simple de razonar.

El costo es que los consumidores deben diseñarse para entrega al menos una vez. Ese es un compromiso razonable. La entrega exactamente una vez a través de una base de datos y un intermediario sin transacciones distribuidas no es achievable en la práctica, y fingir lo contrario lleva a sistemas que silenciosamente descartan o procesan duplicadamente eventos bajo condiciones de fallo.

Escribe el evento con los datos. Retransmítelo de manera confiable. Haz que los consumidores sean idempotentes. Ese es todo el patrón.

Este artículo es parte del clúster Arquitectura de Aplicaciones en Producción.

Fuentes

• PostgreSQL FOR UPDATE SKIP LOCKED
• PostgreSQL LISTEN/NOTIFY
• Enrutador de eventos de buzón de salida de Debezium
• Soporte LISTEN de lib/pq
• Paquete database/sql