Skip to content
wannacode

Events and queues

In short: An event says "this happened" and goes into a queue from which each subscriber takes it on its own; delivery is at least once, so the consumer must be idempotent.

Problem

The slow email slows down the payment

When a customer pays, the shop does four things: charges the card, reserves the stock in the warehouse, sends a confirmation email and records the payment in accounting. The checkout calls them one after another:

var card = new Service("card", 2);
var warehouse = new Service("warehouse", 3);
var mailer = new Service("email", 25); // slow SMTP: 25 ticks
var ledger = new Service("accounting", 4);


void Pay(int order, int total)
{
    int start = Service.Clock;
    card.Call($"charges {total} €");
    warehouse.Call($"reserve stock for {order}");
    mailer.Call($"confirmation for {order}");
    ledger.Call($"{total} € for {order}");
    int waited = Service.Clock - start;
    Console.WriteLine($"✓ {order}: the customer waits {waited} ticks");
}

Every "service" here is an object with a counter: every call adds its ticks. Ticks are the time in the program, instead of seconds. The email goes through a slow mail server and costs 25 ticks:

class Service(string name, int cost, int failures = 0)
{
    public static int Clock; // the ticks that pass
    public int Failures = failures; // a recorded script: how many times in a row it fails


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

The first payment goes through. Then the email fails once:

Pay(1001, 120);
mailer.Failures = 1; // the email fails once
try
{
    Pay(1002, 80);
}
catch (InvalidOperationException e)
{
    Console.WriteLine($"✗ 1002: {e.Message}");
}
Output
card: charges 120 €
warehouse: reserve stock for 1001
email: confirmation for 1001
accounting: 120 € for 1001
✓ 1001: the customer waits 34 ticks
card: charges 80 €
warehouse: reserve stock for 1002
✗ 1002: email is not responding

The customer waits 34 ticks, 25 of them for an email that could just as well arrive a minute later. And with order 1002 the card is charged and the stock is reserved, but the customer sees an error, because the mail server does not answer. They press "Pay" again.

There is a third pain. The checkout knows everyone who reacts to it. Tomorrow marketing wants loyalty points and the mobile app wants a notification. Every new line is a change in the checkout code.

The idea

An event: this already happened

The checkout should not tell the warehouse "reserve", the email "send" and accounting "record". It is enough to announce what happened: "order 1001 is paid". This is a domain event: a fact in the language of the shop, in the past tense, with a name like OrderPaid. It carries its own id and the data the others need: the order number, the total, the quantity. Once it has happened, it does not change.

The event goes into a message queue. The program that keeps the queues is a broker. Everyone who wants to know about payments subscribes. The broker keeps a separate queue for every subscriber and puts a copy of the event in each. This is publish/subscribe: the checkout publishes, the subscribers receive, and neither side knows about the other.

You saw the idea in Observer: whoever wants to know subscribes. The difference is the queue in the middle. Observer calls the subscribers right away, in the same call, and waits for each of them. Here the checkout leaves the event in the queue and answers the customer. The subscribers take it when they can, like the packers in producer and consumer.

The checkout and the warehouse can now be separate modules, even separate services, as in the previous lesson: they talk only through events.

See it

The broker in memory

A real broker is a separate program (Kafka, RabbitMQ). Here it is a class in memory with a map of queues, so you can see the mechanism. The event:

record OrderPaid(string Id, int Order, int Total, int Qty);
In C# the event is a record: the fields are given in the constructor and do not change.

The broker. A subscriber gives a name and a function that handles the event:

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); // a copy for every subscriber
    }


    // one round: each subscriber gets the first in its queue
    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}: stays in the queue");
                continue;
            }
            // the ack is lost: the event stays and comes again
            if (network.AckLost(name)) continue;
            queue.Dequeue(); // ack: take it out of the queue
        }
        return any;
    }
}

The important part is in Deliver. The subscriber gets the first event in its queue, but the event leaves the queue only when the subscriber finishes without an error. The confirmation "I am done" is called an ack (acknowledgement). If the subscriber throws, there is no ack and the event stays for the next round. A lost ack comes from the class Network: a recorded script of whose ack gets lost. Here it loses nothing; on the next card it does.

class Network(params string[] lost)
{
    readonly List<string> lostAcks = [.. lost]; // a recorded script: whose ack is lost


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

The checkout now only charges the card and publishes:

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


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

Two orders, and the email fails once, as in the problem:

var mailer = new Service("email", 25, 1);
var ledger = new Service("accounting", 4);
var broker = new Broker(new Network());
var warehouse = new Warehouse();
broker.Subscribe("warehouse", warehouse.On);
broker.Subscribe("email", e => mailer.Call($"confirmation for {e.Order}"));
broker.Subscribe("accounting", e => ledger.Call($"{e.Total} € for {e.Order}"));
Pay(1001, 120, 2);
Pay(1002, 80, 1);
while (broker.Deliver()) { }
Output
card: charges 120 €
✓ 1001: the customer waits 2 ticks
card: charges 80 €
✓ 1002: the customer waits 2 ticks
warehouse: 2 pcs for 1001, 8 left
  ✗ email is not responding: stays in the queue
accounting: 120 € for 1001
warehouse: 1 pcs for 1002, 7 left
email: confirmation for 1001
accounting: 80 € for 1002
email: confirmation for 1002

The customer waits 2 ticks for both orders. The email is still slow and still fails, but now that is its own problem: the event stays in its queue and comes again in the next round. The warehouse and accounting never notice.

Look at the order too. The confirmation for 1001 goes out after the warehouse has handled 1002. Every queue moves at its own pace. So a subscriber must not count on another subscriber having done its part already.

The idea

At least once means sometimes twice

The broker removes the event from the queue only after the ack. What happens if the warehouse takes out the stock, but the ack gets lost on the network? The broker cannot tell "the warehouse failed before it started" from "the warehouse finished, but could not say so". So as not to lose an order, it delivers the event again.

The script loses the ack from the warehouse once. The warehouse is naive: every event takes out the stock:

class Warehouse
{
    public int Stock = 10;


    public void On(OrderPaid e)
    {
        Stock -= e.Qty;
        Console.WriteLine(
            $"warehouse: {e.Qty} pcs for {e.Order}, {Stock} left");
    }
}
var mailer = new Service("email", 25, 0);
var ledger = new Service("accounting", 4);
var broker = new Broker(new Network("warehouse"));
var warehouse = new Warehouse();
broker.Subscribe("warehouse", warehouse.On);
broker.Subscribe("email", e => mailer.Call($"confirmation for {e.Order}"));
broker.Subscribe("accounting", e => ledger.Call($"{e.Total} € for {e.Order}"));
Pay(1001, 120, 2);
while (broker.Deliver()) { }
Output
card: charges 120 €
✓ 1001: the customer waits 2 ticks
warehouse: 2 pcs for 1001, 8 left
  ✗ ack from warehouse is lost
email: confirmation for 1001
accounting: 120 € for 1001
warehouse: 2 pcs for 1001, 6 left

One order for 2 items, and 4 are missing from the warehouse. The stock is wrong, and the shop will refuse the next order even though the goods are on the shelf.

A broker can promise one of two things:

  • at most once: it removes the event before the subscriber works. No duplicates, but if the subscriber fails, the event is lost;
  • at least once: it removes it after the ack. Nothing is lost, but sometimes it arrives twice.

"Exactly once" across a network is something a broker cannot promise on its own. That is why almost every system picks "at least once", and the receiver catches the duplicate.

See it

An idempotent consumer

A consumer for whom getting the same event a second time changes nothing is idempotent (an idempotent consumer). The warehouse remembers the ids of the events it has already handled and drops the duplicate:

class Warehouse
{
    // the ids of the events already handled
    readonly HashSet<string> seen = [];
    public int Stock = 10;


    public void On(OrderPaid e)
    {
        if (seen.Contains(e.Id))
        {
            Console.WriteLine($"warehouse: {e.Id} is a duplicate");
            return;
        }
        Stock -= e.Qty;
        seen.Add(e.Id);
        Console.WriteLine(
            $"warehouse: {e.Qty} pcs for {e.Order}, {Stock} left");
    }
}

public void Publish(OrderPaid e)
{
    foreach (var queue in queues.Values)
        queue.Enqueue(e); // a copy for every subscriber
}


// one round: each subscriber gets the first in its queue
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}: stays in the queue");
            continue;
        }
        // the ack is lost: the event stays and comes again
        if (network.AckLost(name)) continue;
        queue.Dequeue(); // ack: take it out of the queue
    }
    return any;
}
Step by step, in words
  1. The checkout charges the card and publishes OrderPaid evt-1001: "order 1001 is paid". The customer waits only 2 ticks.
  2. The broker puts a copy in each queue: warehouse, email and accounting. The checkout has already answered.
  3. In each round every subscriber gets the first event in its own queue. The event stays in the queue until the ack.
  4. The warehouse takes out 2 (8 left) and writes evt-1001 in seen. Email and accounting do their part.
  5. The acks from email and accounting arrive, and the event leaves their queues. The ack from the warehouse is lost on the way.
  6. The broker does not know the warehouse did the work, so it delivers evt-1001 again. This is "at least once".
  7. evt-1001 is already in seen, so the warehouse drops it. The stock stays 8, not 6.
  8. This time the ack arrives and the queue is empty. Delivered at least once, handled exactly once.
var mailer = new Service("email", 25, 0);
var ledger = new Service("accounting", 4);
var broker = new Broker(new Network("warehouse"));
var warehouse = new Warehouse();
broker.Subscribe("warehouse", warehouse.On);
broker.Subscribe("email", e => mailer.Call($"confirmation for {e.Order}"));
broker.Subscribe("accounting", e => ledger.Call($"{e.Total} € for {e.Order}"));
Pay(1001, 120, 2);
while (broker.Deliver()) { }
Output
card: charges 120 €
✓ 1001: the customer waits 2 ticks
warehouse: 2 pcs for 1001, 8 left
  ✗ ack from warehouse is lost
email: confirmation for 1001
accounting: 120 € for 1001
warehouse: evt-1001 is a duplicate

The broker still delivers twice, but the warehouse takes out the stock once. The id is written after the work: if the work fails halfway, the event comes again and is handled.

Some operations are idempotent on their own: "the status of 1001 is paid" is the same the second time. "Stock minus 2" is not. Those need an id. In a real system seen is a table in the same database as the stock, and the subtraction and the id are written in one transaction. Otherwise a crash between the two gives a wrong result again. What happens when the database write and sending the event are in different systems is in the next lesson.

Try it

Predict the output

The warehouse is idempotent. This time the script loses the ack from the email. What does the program print?

var mailer = new Service("email", 25, 0);
var ledger = new Service("accounting", 4);
var broker = new Broker(new Network("email"));
var warehouse = new Warehouse();
broker.Subscribe("warehouse", warehouse.On);
broker.Subscribe("email", e => mailer.Call($"confirmation for {e.Order}"));
broker.Subscribe("accounting", e => ledger.Call($"{e.Total} € for {e.Order}"));
Pay(1001, 120, 2);
while (broker.Deliver()) { }
Answer

Try it

Find the bug

A colleague rewrites the warehouse, and the test with the lost ack still shows the stock taken out twice. Which line has the bug?

Pick the line with the bug.

The idea

When events hurt

Events are not free. Do not use them when:

  • you need the answer right now. "Is it in stock?" before the payment is a question, not an event. If you cannot go on without the answer, call the other module;
  • the system is small and in one process. A call or Observer is enough. A broker is one more program that has to run, be watched and be updated.

And know the price when you do use them:

  • the flow is no longer in one place. What happens after the payment cannot be read in one function; you collect it from the subscribers. That is why every event carries an id you can later search for in the logs;
  • the others learn later. The customer pays and opens "My orders" right away, and the invoice is not there yet. This is called eventual consistency. The screen should say so honestly: "the invoice arrives within a minute";
  • order is not promised. You saw it in the program: the queues move at their own pace. If order matters (paid, then cancelled), the events of one order must go through one queue and one consumer;
  • duplicates are your job. Every subscriber must be idempotent.

How a real broker is run and watched in production (a cluster, disks, monitoring) is a topic for the DevOps course ZeroToOps.

More precisely, if you are curious

Real brokers. Kafka keeps the events as a log that each subscriber reads from its own position. There the ack is the remembered position (the offset). RabbitMQ has queues and an ack for every message, as in the lesson. There are also Redis Streams, Amazon SQS and SNS, Azure Service Bus and Google Pub/Sub. In Spring a subscriber is a method with @KafkaListener or @RabbitListener. ASP.NET Core projects often use MassTransit. Python has clients like confluent-kafka and aio-pika, and Celery is a task queue. Node.js has kafkajs and amqplib, and NestJS has a messaging module.

The dead letter queue. An event that fails every time (broken data) must not spin in the queue forever. After a few attempts the broker moves it to a separate queue where a person looks at it.

Exactly once. Kafka has an "exactly once" mode, but it holds inside Kafka: read, process, write back to Kafka. As soon as the effect is outside (an email, a payment, a database), the duplicate is still your job.

Event or command. OrderPaid is an event: many subscribers, none obliged. ReserveStock is a command: one receiver who must carry it out. Both can go through a queue, but the names state different intentions.

How it is in the languages. In C# a subscriber is an Action<OrderPaid>, and you pass a method (warehouse.On). Queue<T> is the queue. foreach (var (name, queue) in queues) takes the key-value pair apart.

Exercise

A notification in the app

Solve it in your editor.

The mobile app wants an "order paid" notification. Add a fourth subscriber, "app", without touching the checkout code. It must be idempotent, like the warehouse. The script loses its ack once:

Output
card: charges 120 €
✓ 1001: the customer waits 2 ticks
warehouse: 2 pcs for 1001, 8 left
email: confirmation for 1001
accounting: 120 € for 1001
app: notification for 1001
  ✗ ack from app is lost
app: evt-1001 is a duplicate

More tasks

  • A cancelled order. Add an OrderCancelled event. The warehouse returns the stock, but only if it has handled OrderPaid for the same order. What happens if the cancellation arrives before the payment?
  • A dead letter queue. If a subscriber fails three times in a row on the same event, the broker moves it to a dead list and goes on with the next one.
  • Loyalty points. Add a subscriber that gives 1 point for every 10 € of the total. What must it remember so as not to give the points twice?

Test

Check yourself

1. Which name is a domain event and not a command?
2. The ack from the warehouse is lost on the way. What does the broker do?
3. Why does the warehouse write the id into seen AFTER it takes out the stock, and not before?
4. When is it better to call the other module directly instead of publishing an event?

Made by Ivo Gergov as part of Evo Tech's campaign for accessible programming education ·