Ir al contenido
wannacode

Eventos y colas

En resumen: Un evento dice «esto pasó» y va a una cola de la que cada suscriptor lo toma por su cuenta; la entrega es al menos una vez, así que el consumidor tiene que ser idempotente.

Problema

El correo lento frena el pago

Cuando un cliente paga, la tienda hace cuatro cosas: cobra la tarjeta, reserva el stock en el almacén, envía un correo de confirmación y anota el pago en la contabilidad. El cobro las llama una tras otra:

var card = new Service("tarjeta", 2);
var warehouse = new Service("almacén", 3);
var mailer = new Service("correo", 25); // SMTP lento: 25 ticks
var ledger = new Service("contabilidad", 4);


void Pay(int order, int total)
{
    int start = Service.Clock;
    card.Call($"cobra {total} €");
    warehouse.Call($"reserva stock para {order}");
    mailer.Call($"confirmación del pedido {order}");
    ledger.Call($"{total} € para {order}");
    int waited = Service.Clock - start;
    Console.WriteLine($"✓ {order}: el cliente espera {waited} ticks");
}

Cada «servicio» aquí es un objeto con un contador: cada llamada suma sus ticks. Los ticks son el tiempo del programa, en vez de segundos. El correo pasa por un servidor de correo lento y cuesta 25 ticks:

class Service(string name, int cost, int failures = 0)
{
    public static int Clock; // los ticks que pasan
    public int Failures = failures; // escenario grabado: cuántas veces seguidas falla


    public void Call(string what)
    {
        Clock += cost;
        if (Failures > 0)
        {
            Failures--;
            throw new InvalidOperationException($"{name} no responde");
        }
        Console.WriteLine($"{name}: {what}");
    }
}

El primer pago sale bien. Luego el correo falla una vez:

Pay(1001, 120);
mailer.Failures = 1; // el correo falla una vez
try
{
    Pay(1002, 80);
}
catch (InvalidOperationException e)
{
    Console.WriteLine($"✗ 1002: {e.Message}");
}
Salida
tarjeta: cobra 120 €
almacén: reserva stock para 1001
correo: confirmación del pedido 1001
contabilidad: 120 € para 1001
✓ 1001: el cliente espera 34 ticks
tarjeta: cobra 80 €
almacén: reserva stock para 1002
✗ 1002: correo no responde

El cliente espera 34 ticks, 25 de ellos por un correo que podría llegar perfectamente un minuto después. Y con el pedido 1002 la tarjeta está cobrada y el stock reservado, pero el cliente ve un error, porque el servidor de correo no responde. Pulsa «Pagar» otra vez.

Hay un tercer dolor. El cobro conoce a todos los que reaccionan. Mañana marketing quiere puntos de fidelidad y la app móvil quiere un aviso. Cada línea nueva es un cambio en el código del cobro.

La idea

Un evento: esto ya pasó

El cobro no debería decirle al almacén «reserva», al correo «envía» y a la contabilidad «anota». Basta con anunciar lo que pasó: «el pedido 1001 está pagado». Esto es un evento de dominio: un hecho en el lenguaje de la tienda, en pasado, con un nombre como OrderPaid. Lleva su propio id y los datos que necesitan los demás: el número de pedido, el total, la cantidad. Lo que ya pasó no cambia.

El evento va a una cola de mensajes. El programa que guarda las colas es un broker. Quien quiera enterarse de los pagos se suscribe. El broker guarda una cola aparte para cada suscriptor y pone una copia del evento en cada una. Esto es publish/subscribe: el cobro publica, los suscriptores reciben, y ninguno de los lados sabe del otro.

Ya viste la idea en Observer: quien quiere enterarse se suscribe. La diferencia es la cola en medio. Observer llama a los suscriptores enseguida, en la misma llamada, y espera a cada uno. Aquí el cobro deja el evento en la cola y le responde al cliente. Los suscriptores lo toman cuando pueden, como los empacadores en productor y consumidor.

El cobro y el almacén ya pueden ser módulos separados, incluso servicios separados, como en la lección anterior: hablan solo mediante eventos.

Míralo

El broker en memoria

Un broker real es un programa aparte (Kafka, RabbitMQ). Aquí es una clase en memoria con un mapa de colas, para que veas el mecanismo. El evento:

record OrderPaid(string Id, int Order, int Total, int Qty);
En C# el evento es un record: los campos se pasan en el constructor y no cambian.

El broker. Un suscriptor da un nombre y una función que procesa el evento:

class Broker(Network network)
{
    readonly Dictionary<string, Queue<OrderPaid>> queues = new();
    readonly Dictionary<string, Action<OrderPaid>> handlers = new();


    public void Subscribe(string name, Action<OrderPaid> handler)
    {
        queues[name] = new Queue<OrderPaid>();
        handlers[name] = handler;
    }


    public void Publish(OrderPaid e)
    {
        foreach (var queue in queues.Values)
            queue.Enqueue(e); // una copia para cada suscriptor
    }


    // una ronda: cada suscriptor recibe el primero de su cola
    public bool Deliver()
    {
        bool any = false;
        foreach (var (name, queue) in queues)
        {
            if (queue.Count == 0) continue;
            any = true;
            try { handlers[name](queue.Peek()); }
            catch (InvalidOperationException e)
            {
                Console.WriteLine($"  ✗ {e.Message}: sigue en la cola");
                continue;
            }
            // el ack se pierde: el evento se queda y llegará otra vez
            if (network.AckLost(name)) continue;
            queue.Dequeue(); // ack: sácalo de la cola
        }
        return any;
    }
}

Lo importante está en Deliver. El suscriptor recibe el primer evento de su cola, pero el evento sale de la cola solo cuando el suscriptor termina sin error. La confirmación «ya terminé» se llama ack (acknowledgement). Si el suscriptor lanza una excepción, no hay ack y el evento se queda para la ronda siguiente. Un ack perdido viene de la clase Network: un escenario grabado de a quién se le pierde el ack. Aquí no pierde nada; en la tarjeta siguiente sí.

class Network(params string[] lost)
{
    readonly List<string> lostAcks = [.. lost]; // escenario grabado: de quién se pierde el ack


    public bool AckLost(string name)
    {
        if (!lostAcks.Remove(name)) return false;
        Console.WriteLine($"  ✗ ack de {name} se pierde");
        return true;
    }
}

El cobro ahora solo cobra la tarjeta y publica:

var card = new Service("tarjeta", 2);


void Pay(int order, int total, int qty)
{
    int start = Service.Clock;
    card.Call($"cobra {total} €");
    broker.Publish(new OrderPaid($"evt-{order}", order, total, qty));
    int waited = Service.Clock - start;
    Console.WriteLine($"✓ {order}: el cliente espera {waited} ticks");
}

Dos pedidos, y el correo falla una vez, como en el problema:

var mailer = new Service("correo", 25, 1);
var ledger = new Service("contabilidad", 4);
var broker = new Broker(new Network());
var warehouse = new Warehouse();
broker.Subscribe("almacén", warehouse.On);
broker.Subscribe("correo", e => mailer.Call($"confirmación del pedido {e.Order}"));
broker.Subscribe("contabilidad", e => ledger.Call($"{e.Total} € para {e.Order}"));
Pay(1001, 120, 2);
Pay(1002, 80, 1);
while (broker.Deliver()) { }
Salida
tarjeta: cobra 120 €
✓ 1001: el cliente espera 2 ticks
tarjeta: cobra 80 €
✓ 1002: el cliente espera 2 ticks
almacén: 2 unidades para 1001, quedan 8
  ✗ correo no responde: sigue en la cola
contabilidad: 120 € para 1001
almacén: 1 unidades para 1002, quedan 7
correo: confirmación del pedido 1001
contabilidad: 80 € para 1002
correo: confirmación del pedido 1002

El cliente espera 2 ticks en los dos pedidos. El correo sigue lento y sigue fallando, pero ahora es problema suyo: el evento se queda en su cola y vuelve en la ronda siguiente. El almacén y la contabilidad ni se enteran.

Fíjate también en el orden. La confirmación del 1001 sale después de que el almacén procesó el 1002. Cada cola avanza a su ritmo. Por eso un suscriptor no puede contar con que otro ya hizo su parte.

La idea

Al menos una vez significa a veces dos

El broker quita el evento de la cola solo después del ack. ¿Qué pasa si el almacén saca el stock, pero el ack se pierde en la red? El broker no distingue «el almacén falló antes de empezar» de «el almacén terminó, pero no pudo decirlo». Para no perder un pedido, entrega el evento otra vez.

El escenario pierde el ack del almacén una vez. El almacén es ingenuo: cada evento saca stock:

class Warehouse
{
    public int Stock = 10;


    public void On(OrderPaid e)
    {
        Stock -= e.Qty;
        Console.WriteLine(
            $"almacén: {e.Qty} unidades para {e.Order}, quedan {Stock}");
    }
}
var mailer = new Service("correo", 25, 0);
var ledger = new Service("contabilidad", 4);
var broker = new Broker(new Network("almacén"));
var warehouse = new Warehouse();
broker.Subscribe("almacén", warehouse.On);
broker.Subscribe("correo", e => mailer.Call($"confirmación del pedido {e.Order}"));
broker.Subscribe("contabilidad", e => ledger.Call($"{e.Total} € para {e.Order}"));
Pay(1001, 120, 2);
while (broker.Deliver()) { }
Salida
tarjeta: cobra 120 €
✓ 1001: el cliente espera 2 ticks
almacén: 2 unidades para 1001, quedan 8
  ✗ ack de almacén se pierde
correo: confirmación del pedido 1001
contabilidad: 120 € para 1001
almacén: 2 unidades para 1001, quedan 6

Un pedido de 2 unidades, y al almacén le faltan 4. El stock está mal, y la tienda rechazará el pedido siguiente aunque la mercancía esté en el estante.

Un broker puede prometer una de dos cosas:

  • como máximo una vez: quita el evento antes de que el suscriptor trabaje. Sin duplicados, pero si el suscriptor falla, el evento se pierde;
  • al menos una vez: lo quita después del ack. No se pierde nada, pero a veces llega dos veces.

«Exactamente una vez» a través de una red es algo que un broker no puede prometer por sí solo. Por eso casi todos los sistemas eligen «al menos una vez», y el receptor atrapa el duplicado.

Míralo

Un consumidor idempotente

Un consumidor para quien recibir el mismo evento por segunda vez no cambia nada es idempotente (idempotent consumer). El almacén recuerda los ids de los eventos que ya procesó y descarta el duplicado:

class Warehouse
{
    // los ids de los eventos ya procesados
    readonly HashSet<string> seen = [];
    public int Stock = 10;


    public void On(OrderPaid e)
    {
        if (seen.Contains(e.Id))
        {
            Console.WriteLine($"almacén: {e.Id} es un duplicado");
            return;
        }
        Stock -= e.Qty;
        seen.Add(e.Id);
        Console.WriteLine(
            $"almacén: {e.Qty} unidades para {e.Order}, quedan {Stock}");
    }
}

public void Publish(OrderPaid e)
{
    foreach (var queue in queues.Values)
        queue.Enqueue(e); // una copia para cada suscriptor
}


// una ronda: cada suscriptor recibe el primero de su cola
public bool Deliver()
{
    bool any = false;
    foreach (var (name, queue) in queues)
    {
        if (queue.Count == 0) continue;
        any = true;
        try { handlers[name](queue.Peek()); }
        catch (InvalidOperationException e)
        {
            Console.WriteLine($"  ✗ {e.Message}: sigue en la cola");
            continue;
        }
        // el ack se pierde: el evento se queda y llegará otra vez
        if (network.AckLost(name)) continue;
        queue.Dequeue(); // ack: sácalo de la cola
    }
    return any;
}
Paso a paso, en palabras
  1. El cobro carga la tarjeta y publica OrderPaid evt-1001: «el pedido 1001 está pagado». El cliente espera solo 2 ticks.
  2. El broker pone una copia en cada cola: almacén, correo y contabilidad. El cobro ya ha respondido.
  3. En cada ronda cada suscriptor recibe el primer evento de su propia cola. El evento sigue en la cola hasta el ack.
  4. El almacén saca 2 unidades (quedan 8) y anota evt-1001 en seen. El correo y la contabilidad hacen lo suyo.
  5. Los acks del correo y de la contabilidad llegan, y el evento sale de sus colas. El ack del almacén se pierde por el camino.
  6. El broker no sabe que el almacén ya hizo el trabajo y entrega evt-1001 otra vez. Esto es «al menos una vez».
  7. evt-1001 ya está en seen, así que el almacén lo descarta. El stock se queda en 8, no en 6.
  8. Esta vez el ack llega y la cola queda vacía. Entregado al menos una vez, procesado exactamente una vez.
var mailer = new Service("correo", 25, 0);
var ledger = new Service("contabilidad", 4);
var broker = new Broker(new Network("almacén"));
var warehouse = new Warehouse();
broker.Subscribe("almacén", warehouse.On);
broker.Subscribe("correo", e => mailer.Call($"confirmación del pedido {e.Order}"));
broker.Subscribe("contabilidad", e => ledger.Call($"{e.Total} € para {e.Order}"));
Pay(1001, 120, 2);
while (broker.Deliver()) { }
Salida
tarjeta: cobra 120 €
✓ 1001: el cliente espera 2 ticks
almacén: 2 unidades para 1001, quedan 8
  ✗ ack de almacén se pierde
correo: confirmación del pedido 1001
contabilidad: 120 € para 1001
almacén: evt-1001 es un duplicado

El broker sigue entregando dos veces, pero el almacén saca el stock una sola vez. El id se escribe después del trabajo: si el trabajo falla a medias, el evento vuelve y se procesa.

Algunas operaciones son idempotentes por sí solas: «el estado de 1001 es pagado» es lo mismo la segunda vez. «Stock menos 2» no. Para esas hace falta un id. En un sistema real seen es una tabla en la misma base de datos que el stock, y la resta y el id se escriben en una sola transacción. Si no, una caída entre las dos vuelve a dar un resultado incorrecto. Qué pasa cuando la escritura en la base de datos y el envío del evento están en sistemas distintos se cuenta en la lección siguiente.

Pruébalo

Predice la salida

El almacén es idempotente. Esta vez el escenario pierde el ack del correo. ¿Qué imprime el programa?

var mailer = new Service("correo", 25, 0);
var ledger = new Service("contabilidad", 4);
var broker = new Broker(new Network("correo"));
var warehouse = new Warehouse();
broker.Subscribe("almacén", warehouse.On);
broker.Subscribe("correo", e => mailer.Call($"confirmación del pedido {e.Order}"));
broker.Subscribe("contabilidad", e => ledger.Call($"{e.Total} € para {e.Order}"));
Pay(1001, 120, 2);
while (broker.Deliver()) { }
Respuesta

Pruébalo

Encuentra el error

Un colega reescribe el almacén, y la prueba con el ack perdido sigue mostrando el stock sacado dos veces. ¿Qué línea tiene el error?

Elige la línea con el error.

La idea

Cuándo los eventos hacen daño

Los eventos no son gratis. No los uses cuando:

  • necesitas la respuesta ahora mismo. «¿Hay stock?» antes del pago es una pregunta, no un evento. Si no puedes seguir sin la respuesta, llama al otro módulo;
  • el sistema es pequeño y está en un solo proceso. Basta una llamada u Observer. Un broker es un programa más que hay que mantener en marcha, vigilar y actualizar.

Y conoce el precio cuando sí los uses:

  • el flujo ya no está en un solo lugar. Lo que pasa después del pago no se lee en una sola función; lo reúnes desde los suscriptores. Por eso cada evento lleva un id que luego puedes buscar en los logs;
  • los demás se enteran después. El cliente paga y abre enseguida «Mis pedidos», y la factura todavía no está. Esto se llama consistencia eventual. La pantalla debería decirlo con honestidad: «la factura llega en un minuto»;
  • no se promete un orden. Lo viste en el programa: las colas avanzan a su ritmo. Si el orden importa (pagado, luego cancelado), los eventos de un pedido deben pasar por una sola cola y un solo consumidor;
  • los duplicados son cosa tuya. Todos los suscriptores tienen que ser idempotentes.

Cómo se opera y se vigila un broker real en producción (un clúster, discos, monitoreo) es tema del curso de DevOps ZeroToOps.

Más precisamente, para curiosos

Brokers reales. Kafka guarda los eventos como un log que cada suscriptor lee desde su propia posición. Allí el ack es la posición recordada (el offset). RabbitMQ tiene colas y un ack para cada mensaje, como en la lección. También existen Redis Streams, Amazon SQS y SNS, Azure Service Bus y Google Pub/Sub. En Spring un suscriptor es un método con @KafkaListener o @RabbitListener. Los proyectos de ASP.NET Core suelen usar MassTransit. Python tiene clientes como confluent-kafka y aio-pika, y Celery es una cola de tareas. Node.js tiene kafkajs y amqplib, y NestJS tiene un módulo de mensajería.

La dead letter queue. Un evento que falla siempre (datos rotos) no debe dar vueltas en la cola para siempre. Tras unos intentos el broker lo pasa a una cola aparte donde lo mira una persona.

Exactamente una vez. Kafka tiene un modo «exactly once», pero vale dentro de Kafka: leer, procesar, volver a escribir en Kafka. En cuanto el efecto está fuera (un correo, un pago, una base de datos), el duplicado sigue siendo cosa tuya.

Evento u orden. OrderPaid es un evento: muchos suscriptores, ninguno obligado. ReserveStock es una orden: un receptor que debe cumplirla. Ambos pueden pasar por una cola, pero los nombres expresan intenciones distintas.

Cómo es en los lenguajes. En C# un suscriptor es una Action<OrderPaid>, y pasas un método (warehouse.On). Queue<T> es la cola. foreach (var (name, queue) in queues) descompone el par clave-valor.

Ejercicio

Un aviso en la app

Resuelve la tarea en tu editor.

La app móvil quiere un aviso de «pedido pagado». Añade un cuarto suscriptor, «app», sin tocar el código del cobro. Tiene que ser idempotente, como el almacén. El escenario pierde su ack una vez:

Salida
tarjeta: cobra 120 €
✓ 1001: el cliente espera 2 ticks
almacén: 2 unidades para 1001, quedan 8
correo: confirmación del pedido 1001
contabilidad: 120 € para 1001
app: aviso del pedido 1001
  ✗ ack de app se pierde
app: evt-1001 es un duplicado

Más tareas

  • Un pedido cancelado. Añade un evento OrderCancelled. El almacén devuelve el stock, pero solo si procesó OrderPaid del mismo pedido. ¿Qué pasa si la cancelación llega antes que el pago?
  • Una dead letter queue. Si un suscriptor falla tres veces seguidas con el mismo evento, el broker lo pasa a una lista dead y sigue con el siguiente.
  • Puntos de fidelidad. Añade un suscriptor que da 1 punto por cada 10 € del total. ¿Qué tiene que recordar para no dar los puntos dos veces?

Test

Ponte a prueba

1. ¿Qué nombre es un evento de dominio y no una orden?
2. El ack del almacén se pierde por el camino. ¿Qué hace el broker?
3. ¿Por qué el almacén escribe el id en seen DESPUÉS de sacar el stock y no antes?
4. ¿Cuándo es mejor llamar directamente al otro módulo en vez de publicar un evento?

Creado por Ivo Gergov como parte de la campaña de Evo Tech por una educación accesible en programación ·