24Messaging and Event-Driven Architecture

A consumer of the fills topic crashes after applying a batch of messages to the positions and before recording how far it has read. When it restarts it reads the batch again and applies it again, and the firm’s position in one stock is double until someone compares it with the drop copy. Nothing was broken: the messaging system delivered every message at least once, which is exactly what it promised. This chapter is about what messaging systems promise, what they cannot, and how a consumer turns at-least-once delivery into exactly-once effects — with a partitioned log built in process, a position service run under a thousand crashes in three modes, and the outbox that lets a service change its database and announce it without losing either.

24.1 Why systems talk through messages

The platform map of chapter 1 is a set of services: trade capture, positions, risk, P&L, surveillance. They can call each other directly, each one knowing who needs what; or each can publish what happened, and whoever needs it subscribes. The second decouples them in three ways — in knowledge (the publisher does not know its consumers), in time (a consumer can be down and catch up), and in rate (a slow consumer does not slow the publisher) — and makes the flow of events itself a record that can be replayed.

The cost is that the call graph becomes invisible, and failures move from the call (which fails loudly) to the delivery (which can lose or repeat messages quietly). Most of this chapter is about the second.

24.2 Brokers and brokerless transports

The two answer different needs. A broker adds a hop and a server to run, and gives in return storage, replay, and consumers that come and go. Brokerless transports give the lowest latency and no central component: the market-data and order paths of Book 13 are brokerless, with multicast (Book 13, chapter 16) and ring buffers in shared memory, and libraries such as Aeron — "efficient reliable UDP unicast, UDP multicast, and IPC message transport" — provide the reliability on top. Between services that care more about not losing an event than about microseconds — trade capture, positions, risk, the post-trade chain — the firm uses a log.

24.3 The log as the backbone

This is the design Kreps, Narkhede and Rao described for Kafka in 2011: a message has no identifier of its own, "each message is addressed by its logical offset in the log". The chapter’s firm.eventlog builds it in process: topics, partitions by a stable hash of the key, append-only segments on disk, retention that drops whole segments, and committed offsets per group. Keying fills by account puts all of an account’s fills in one partition, in order; eight partitions shared by three consumers are assigned three, three and two.

    def append(self, topic, key, value, pid=None, seq=None) -> tuple[int, int]:
        """Append to the key's partition; a (pid, seq) already stored is ignored."""
        p = self.partition(topic, key)
        if pid is not None:
            if (pid, seq) in self.seen[(topic, p)]:
                return p, -1                           # duplicate send: ignored
            self.seen[(topic, p)].add((pid, seq))
        off = self.end(topic, p)
        self.parts[topic][p].append((off, key, value))
        if self.dir is not None:
            self._write(topic, p, off, key, value)
        return p, off

24.4 Delivery guarantees and the exactly-once myth

A consumer does two things with a batch: it processes the messages, and it commits the offset. A crash can fall between them, and the order of the two decides what the crash costs.

The 2011 paper was explicit: "Kafka only guarantees at-least-once delivery", and an application that cares about duplicates "must add its own deduplication logic". Later versions added an idempotent producer and transactions, which give exactly-once between Kafka topics; for any other destination, the documentation says, exactly-once "generally requires cooperation with such systems". The position service is such a destination.

def run_batch(log, group, topic, p, n, apply, mode, crash=None, store=None) -> int:
    """Read up to n messages from the committed offset and process them under `mode`;
    crash(stage, i) may raise Crash to simulate the process dying at that point."""
    crash = crash or (lambda stage, i: None)
    start = store.offset(topic, p) if mode == "atomic" else log.committed(group, topic, p)
    batch = log.read(topic, p, start, n)
    if not batch:
        return 0
    nxt = batch[-1][0] + 1
    if mode == "at-most-once":
        log.commit(group, topic, p, nxt)             # commit first: a crash loses the rest
    if mode == "atomic":
        store.begin()
    for i, (_off, _key, value) in enumerate(batch):
        crash("process", i)
        apply(value)
    crash("before-commit", len(batch))
    if mode == "at-least-once":
        log.commit(group, topic, p, nxt)             # commit after: a crash repeats it
    if mode == "atomic":
        store.commit(topic, p, nxt)                  # state and offset in one transaction
    return len(batch)

The chapter’s day is 10 000 synthetic fills for ten accounts and twenty stocks, keyed by account into four partitions and consumed in batches of 100. Each day ten crashes strike at points drawn uniformly over the consumer’s steps — every message processed, and the moment between processing a batch and committing it — and a hundred days are run in each mode. For every fill we count how many times it reached the positions.

The first two modes are mirror images (Figure 24.2). A crash inside a batch of 100 falls on average half-way, so committing first loses about 50 fills and committing after repeats about 50; a crash just after processing and before the commit — the hook’s — repeats all 100. In the third mode the positions and the next offset are written in one SQLite transaction: a crash anywhere before the commit rolls both back, the consumer resumes from the offset that matches the state, and no fill is lost or counted twice in 1 000 crashes. The same effect comes from idempotent processing: store each fill’s identifier with the state and skip identifiers already applied.

24.5 Patterns: outbox, event sourcing, dead letters

The consumer’s problem has a mirror on the producer’s side. The trade-capture service must store a trade in its database and publish the event that announces it. Two separate writes cannot be atomic: write then publish, and a crash in between leaves a trade nobody hears about; publish then write, and consumers act on a trade that does not exist.

    def write(self, trade: dict, event: dict) -> None:
        """The trade and the event that announces it, committed together or not at all."""
        self.db.execute("BEGIN")
        self.db.execute("INSERT INTO trades VALUES (?, ?)",
                        (trade["id"], json.dumps(trade)))
        self.db.execute("INSERT INTO outbox (key, body, sent) VALUES (?, ?, 0)",
                        (event["key"], json.dumps(event)))
        self.db.execute("COMMIT")

    def relay(self, log: Log, topic: str, crash=None) -> int:
        """Publish unsent events in order, with the outbox number as idempotence key, then
        mark each sent; a crash in between resends it later, and the log ignores it."""
        n = 0
        rows = self.db.execute("SELECT n, key, body FROM outbox WHERE sent = 0 "
                               "ORDER BY n").fetchall()
        for num, key, body in rows:
            log.append(topic, key, json.loads(body), pid="outbox", seq=num)
            if crash:
                crash("published", num)
            self.db.execute("UPDATE outbox SET sent = 1 WHERE n = ?", (num,))
            n += 1
        return n

Table 24.1 shows the three failure modes and the fix. The outbox moves the problem from two systems to one, and what is left — a relay that published and died before marking — is a repeat, which the idempotent producer or the consumer removes.

Chapter 21’s trade store is event-sourced, and the log is its natural home: a new downstream system subscribes and replays the topic from its start to build its own state. The price is retention — the log must keep every event, or a snapshot with the offset it covers — and discipline about changing the events’ format.

In a partitioned log a message that always fails is worse than in a queue: the consumer cannot skip it without committing past it, so every message behind it in the partition waits. The dead-letter topic lets the consumer move on and leaves a record to be repaired and replayed — which, for fills, must be done before anyone trusts the positions.

24.6 Tutorial: a thousand crashes

24.7 Build: the event log

24.8 Exercises

24.9 Problem: Double for Eleven Minutes

24.10 Interview questions