Patrón Transactional Outbox en Java

Problema

Pagado, y no sale ningún paquete

Desde la lección anterior la tienda publica un evento en cuanto se guarda el pedido. Guardar va a la base de datos, y el evento va al broker. Aquí hay dos maneras de escribirlo. En ellas db es la base de datos, queue es la cola del broker y crash.at es un punto donde el proceso puede morir según un guion grabado:

// ASÍ NO: guardar y el evento están en dos sistemas separados
static void saveThenPublish(int id, int total) {
    db.transaction(tx -> tx.orders.put(id, total));
    crash.at("tras guardar");
    queue.add("evt-" + id + ": OrderPlaced " + id);
}

static void publishThenSave(int id, int total) {
    queue.add("evt-" + id + ": OrderPlaced " + id);
    db.transaction(tx -> {
        tx.orders.put(id, total);
        crash.at("antes del commit");
    });
}

En cada función el proceso muere una vez, en su propio punto:

crash = new CrashScript("tras guardar");
try {
    saveThenPublish(1001, 120);
} catch (Crash e) {
    System.out.println("✗ el proceso cae " + e.getMessage());
}
crash = new CrashScript("antes del commit");
try {
    publishThenSave(1002, 80);
} catch (Crash e) {
    System.out.println("✗ el proceso cae " + e.getMessage());
}
System.out.println("pedidos: " + db.orders.keySet());
System.out.println("cola: " + queue);

✗ el proceso cae tras guardar
✗ el proceso cae antes del commit
pedidos: [1001]
cola: [evt-1002: OrderPlaced 1002]

El pedido 1001 está en la base de datos, pero no hay evento para él. El cliente pagó y el almacén no sabe que tiene que enviar un paquete. El evento de 1002 está en la cola, pero el pedido 1002 no existe: la transacción se revirtió. El almacén empaqueta un pedido que no existe.

Una caída no es rara. El proceso se reinicia en cada deploy, la máquina se queda sin luz, la red hacia el broker se corta un segundo. Con miles de pedidos al día, alguno cae justo entre las dos líneas.

La idea

Dos sistemas, ninguna transacción común

Una transacción promete «todo o nada», pero solo para las escrituras de su propia base de datos. El broker es otro sistema. El commit de la base de datos no sabe nada de él, y un mensaje enviado al broker no se puede «recuperar». Así que, sea cual sea el orden de las dos líneas, una caída entre ellas deja un sistema sin el otro.

Existe un protocolo para una transacción común entre dos sistemas (two-phase commit), pero la mayoría de los brokers no lo soporta, y donde existe es lento y frágil.

La solución es no tener dos sistemas en el momento de guardar. El evento se escribe como una fila en una tabla de la misma base de datos, en la misma transacción que el pedido. La tabla se llama outbox. El commit escribe las dos filas, o ninguna. Luego un relay aparte lee la outbox, envía al broker las filas que aún no se enviaron y las marca. Este es el patrón Transactional Outbox.

Míralo

La base de datos y la caída en memoria

La base de datos aquí es un objeto con dos tablas. La transacción junta las escrituras en Tx y las pasa a las tablas de una vez, solo si el trabajo llega hasta el final. Si lanza una excepción a medias, no se escribe nada: eso es un rollback.

class OutboxRow {
    final String id;
    final String event;
    boolean sent = false;

    OutboxRow(String id, String event) {
        this.id = id;
        this.event = event;
    }
}

// las escrituras de una transacción, antes del commit
class Tx {
    final Map<Integer, Integer> orders = new TreeMap<>();
    final List<OutboxRow> outbox = new ArrayList<>();
}

class Db {
    final Map<Integer, Integer> orders = new TreeMap<>();
    final List<OutboxRow> outbox = new ArrayList<>();

    // todo o nada: las escrituras llegan a las tablas solo si el trabajo llega hasta el final
    void transaction(Consumer<Tx> work) {
        Tx tx = new Tx();
        work.accept(tx);
        orders.putAll(tx.orders); // commit
        outbox.addAll(tx.outbox);
    }
}

El trabajo es una lambda que recibe tx. Si lanza una excepción, las líneas que siguen a su llamada (el commit) no se ejecutan.

La caída es una excepción en un punto exacto. La base de datos y la cola son sistemas separados: sobreviven a la caída, y el programa sigue como un proceso reiniciado.

class Crash extends RuntimeException {
    Crash(String where) {
        super(where);
    }
}

// un guion grabado: dónde cae el proceso (sin punto: en ninguna parte)
class CrashScript {
    private final String point;

    CrashScript(String point) {
        this.point = point;
    }

    CrashScript() {
        this("");
    }

    void at(String here) {
        if (here.equals(point)) throw new Crash(here);
    }
}

static Db db = new Db();
static List<String> queue = new ArrayList<>(); // la cola del broker
static CrashScript crash = new CrashScript();

Cómo se ve lo mismo con una base de datos real lo viste en la lección sobre bases de datos.

Míralo

El pedido y la fila de la outbox juntos

Guardar el pedido también pone una fila en la outbox, en la misma transacción. No envía nada al broker:

static void placeOrder(int id, int total) {
    db.transaction(tx -> {
        tx.orders.put(id, total);
        // en la misma transacción
        tx.outbox.add(new OutboxRow("evt-" + id, "OrderPlaced " + id));
        crash.at("antes del commit");
    });
}

El relay recorre la outbox, envía cada fila que aún no se envió y la marca:

// el relay envía las filas sin enviar de la outbox
static void relay() {
    for (OutboxRow row : db.outbox) {
        if (row.sent) continue;
        queue.add(row.id + ": " + row.event);
        crash.at("en el relay");
        row.sent = true;
    }
}

static void show() {
    System.out.println("pedidos: " + db.orders.keySet());
    System.out.println("outbox: " + db.outbox.stream().map(r -> r.id + (r.sent ? " ✓" : "")).toList());
    System.out.println("cola: " + queue);
}

El pedido 1003 pasa. El pedido 1004 muere antes del commit. Luego arranca el relay, como tras un reinicio:

placeOrder(1003, 120);
show();
crash = new CrashScript("antes del commit");
try {
    placeOrder(1004, 80);
} catch (Crash e) {
    System.out.println("✗ el proceso cae " + e.getMessage());
}
crash = new CrashScript();
relay(); // tras el reinicio
show();

pedidos: [1003]
outbox: [evt-1003]
cola: []
✗ el proceso cae antes del commit
pedidos: [1003]
outbox: [evt-1003 ✓]
cola: [evt-1003: OrderPlaced 1003]

static void placeOrder(int id, int total) {
    db.transaction(tx -> {
        tx.orders.put(id, total);
        // en la misma transacción
        tx.outbox.add(new OutboxRow("evt-" + id, "OrderPlaced " + id));
        crash.at("antes del commit");
    });
}

// ASÍ NO: guardar y el evento están en dos sistemas separados
static void saveThenPublish(int id, int total) {
    db.transaction(tx -> tx.orders.put(id, total));
    crash.at("tras guardar");
    queue.add("evt-" + id + ": OrderPlaced " + id);
}

static void publishThenSave(int id, int total) {
    queue.add("evt-" + id + ": OrderPlaced " + id);
    db.transaction(tx -> {
        tx.orders.put(id, total);
        crash.at("antes del commit");
    });
}

static void placeOrder(int id, int total) {
    db.transaction(tx -> {
        tx.orders.put(id, total);
        // en la misma transacción
        tx.outbox.add(new OutboxRow("evt-" + id, "OrderPlaced " + id));
        crash.at("antes del commit");
    });
}

// el relay envía las filas sin enviar de la outbox
static void relay() {
    for (OutboxRow row : db.outbox) {
        if (row.sent) continue;
        queue.add(row.id + ": " + row.event);
        crash.at("en el relay");
        row.sent = true;
    }
}

Paso a paso, en palabras

Escenario 1: primero guardar, luego el evento. El pedido 1001 entra en la base de datos con un commit.
El proceso cae justo entre las dos líneas. El pedido está en la base de datos, pero el evento no sale nunca.
Esta línea no se ejecuta nunca. El almacén no se entera de 1001: el cliente pagó y ningún paquete saldrá.
Escenario 2: al revés. El evento evt-1002 sale primero.
Luego el pedido se escribe en una transacción.
El proceso cae antes del commit y la base de datos descarta la escritura. El pedido 1002 no existe.
Pero el evento ya está en la cola: el almacén empaqueta un pedido que no existe.
Escenario 3, la outbox: el pedido 1003 y su evento, como una fila en la tabla outbox, en una transacción.
Commit: las dos filas entran a la vez. O las dos, o ninguna.
El proceso cae antes de enviar nada. No se pierde nada: la fila espera en la outbox.
Tras el reinicio el relay lee la outbox y encuentra la fila sin enviar.
Envía evt-1003 a la cola y el almacén lo recibe.
Solo entonces marca la fila. Una caída entre las dos envía la fila otra vez, por eso el consumidor debe ser idempotente.
Guardar y el evento ya no se separan: los dos llegan, tarde o temprano.

Tras el commit el pedido y la fila de la outbox están en la base de datos, y la cola sigue vacía: el proceso puede morir aquí y no se pierde nada. El pedido 1004 no deja nada: ni pedido ni fila, porque los dos están en una transacción. El relay encuentra la fila sin enviar y la envía.

En un sistema real el relay es un bucle que pregunta a la base de datos cada segundo: «dame las filas que aún no se enviaron». Por eso el evento sale un poco después del commit, no en el mismo instante.

Pruébalo

Predice la salida

El relay envía la fila y muere antes de marcarla. Tras el reinicio corre otra vez. ¿Qué hay en la cola?

La última línea de la salida es:

placeOrder(1003, 120);
crash = new CrashScript("en el relay");
try {
    relay();
} catch (Crash e) {
    System.out.println("✗ el proceso cae " + e.getMessage());
}
crash = new CrashScript();
relay(); // tras el reinicio
show();

Respuesta

cola: [evt-1003: OrderPlaced 1003]

cola: [evt-1003: OrderPlaced 1003, evt-1003: OrderPlaced 1003]

cola: []

Pruébalo

Encuentra el error

Un compañero decide que los duplicados molestan e intercambia dos líneas del relay. La caída es en el mismo punto, entre las dos líneas:

✗ el proceso cae en el relay
pedidos: [1003]
outbox: [evt-1003 ✓]
cola: []

La fila de la outbox está marcada y la cola está vacía. ¿Qué línea tiene el error?

Elige la línea con el error.

Pruébalo

Pon las líneas en orden

Pon el cuerpo del bucle del relay en un orden en el que no se pierda nada.

row.sent = true;
queue.add(row.id + ": " + row.event);
if (row.sent) continue;

La idea

Cuándo una outbox no vale la pena

Una outbox añade una tabla, un relay y un retraso. No la uses cuando:

no hay un broker aparte. Si las reacciones están en el mismo proceso y escriben en la misma base de datos (Observer en la misma transacción), basta una transacción;
perder un mensaje no hace daño. Actualizar estadísticas o calentar una caché puede perderse una vez al mes sin que nadie salga perjudicado;
el evento debe salir en el mismo milisegundo. El relay lee a intervalos. Un segundo de retraso suele estar bien, pero no siempre.

Y cuando la uses, piensa en:

duplicados. El relay envía al menos una vez. Los consumidores deben ser idempotentes;
orden. Un relay envía en el orden de la outbox. Varios relays a la vez pueden alterar el orden o enviar una fila dos veces, por eso las filas se bloquean al leerlas;
crecimiento. Las filas enviadas se borran tras unos días, si no la tabla crece para siempre.

Del otro lado hay un patrón espejo, la inbox: el consumidor escribe el id del evento recibido en su propia base de datos, en la misma transacción que su trabajo. Es el seen del almacén de la lección anterior, pero en una tabla.

Más precisamente, para curiosos

El relay en un sistema real. El más simple es un bucle con la consulta SELECT … FROM outbox WHERE sent = false ORDER BY id LIMIT 100. En PostgreSQL FOR UPDATE SKIP LOCKED permite que varios relays corran sin tomar la misma fila. Otra vía es change data capture: Debezium lee el log de la base de datos (el WAL en PostgreSQL, el binlog en MySQL) y envía las filas nuevas de la outbox a Kafka, sin consultas a intervalos. Cómo se ejecuta y se monitorea Debezium o Kafka Connect es un tema del curso de DevOps ZeroToOps.

Soluciones ya hechas. Spring Modulith tiene un event publication registry, que es una outbox. En .NET, MassTransit y NServiceBus tienen una outbox integrada sobre Entity Framework. En Python y Node.js suele escribirse a mano, con una tabla y un relay pequeño, como en esta lección.

Two-phase commit. Las transacciones XA (JTA en Java, System.Transactions en .NET) prometen una transacción común entre una base de datos y un broker. En la práctica Kafka no las soporta, ni RabbitMQ ni las colas de la nube. Por eso la outbox es la respuesta habitual.

¿Por qué no reintentar sin más? «Si el envío falla, inténtalo otra vez» ayuda con un error del broker, pero no con una caída del proceso: tras el reinicio nadie recuerda que algo tenía que salir. La outbox es la memoria que sobrevive a la caída.

Ejercicio

Un relay por lotes

Resuélvelo en tu propio editor.

En un sistema real el relay lee como máximo unas pocas filas a la vez (LIMIT en la consulta). Escribe relay(limit), que envía como máximo limit filas aún sin enviar y devuelve cuántas envió. Guarda tres pedidos y corre el relay con limit 2 hasta que devuelva 0:

ronda 1: enviados 2
ronda 2: enviados 1
ronda 3: enviados 0
cola: 3

Más tareas

Un pedido cancelado. Añade cancelOrder(id), que en una transacción quita el pedido y pone OrderCancelled en la outbox. Comprueba que una caída antes del commit no quita el pedido y no envía nada.
Limpieza. Añade a la fila el momento del envío (un tick) y una función que borre las filas enviadas con más de 10 ticks.
Inbox. Escribe un almacén que guarde los ids de los eventos recibidos y descarte el duplicado de la tarjeta «Predice la salida».

Test

Ponte a prueba

1. ¿Por qué la transacción de la base de datos no protege también el envío al broker?Porque las transacciones son de solo lecturaEl broker es otro sistema: el commit de la base de datos no sabe nada de él, y al revésSí lo protege, si el envío está dentro del bloque de la transacción

2. ¿Qué escribe la tienda en una transacción cuando usa una outbox?El pedido y una fila en la tabla outbox con el eventoEl pedido, y envía el evento justo después del commitSolo el evento, y el relay escribe el pedido

3. El relay envía una fila y muere antes de marcarla. ¿Qué pasa tras el reinicio?Envía el mismo evento otra vez: la cola tiene un duplicado y el consumidor debe ser idempotenteEl evento se pierdeNada: el broker recuerda que lo recibió

4. ¿Cuándo sobra una outbox?Siempre, cuando hay una base de datosCuando no hay un broker aparte: los suscriptores están en el mismo proceso y trabajan en la misma transacciónCuando hay más de cien pedidos al día

Transactional Outbox

Cookies