Commentary: pubsub-broker

One publisher, one topic, and a subscriber that consumes slower than the publisher produces. Its queue fills, and now the broker has to throw something away. There are two obvious policies for choosing what — drop the new arrival, or evict the oldest queued message — and the interesting fact is that they lose exactly the same number of messages, 791 of 1000, in every configuration tested. The overflow policy does not decide how much you lose. A study guide for broker.py.

cd pubsub-broker
python3 demo.py           # the aha (§6), instant
python3 test_broker.py    # pins every number this page claims

1. Orientation

This toy is a publish/subscribe broker: publishers send to a topic, subscribers receive everything on the topics they subscribed to, and neither knows the other exists. The mechanism it teaches is what happens at the point where that decoupling meets a finite amount of memory.

Every subscriber gets its own bounded queue. That is the design that delivers the property pub/sub is sold on — one slow consumer falls behind by itself, without holding up the fast ones — and it is also what forces the question this page is about, because a private queue is a bounded queue, and a bounded queue eventually has to refuse something.

Three ways to refuse, all implemented here:

• drop-newest — reject the arrival; the queue keeps what it already had.
• drop-oldest — evict the head to make room; the queue keeps the freshest.
• block — refuse the publish, for every subscriber on the topic.

The shortest version of the result:

from broker import DROP_NEWEST, DROP_OLDEST, run

for policy in (DROP_NEWEST, DROP_OLDEST):
    broker, subs = run(policy, ticks=200, publish_rate=5, capacity=10,
                       drain_rates=(1, 5))
    print(policy, "lost", subs[0].dropped, "of", broker.published)
# drop-newest lost 791 of 1000
# drop-oldest lost 791 of 1000

By the end you should be able to say why those two numbers are equal by construction rather than by coincidence, why adding buffer capacity buys a fixed number of messages rather than a lower loss rate, and what backpressure actually costs the subscribers that were keeping up fine.

2. The problem this mechanism exists to solve

A broker sits between producers and consumers that run at different speeds, and its whole value is that neither has to know about the other. That works as long as the consumer keeps up on average. When it doesn't, the broker is holding messages that have been produced and not yet consumed, and it has a finite amount of memory to hold them in.

The competing goals that make more than one design defensible:

• Don't lose anything. The publisher was told the message was accepted; losing it silently is the worst outcome for an audit log or a payment event.
• Don't let one slow consumer stall the system. If a misbehaving subscriber can stop the publisher, then every subscriber's availability depends on the worst subscriber, and the decoupling pub/sub promised is gone.
• Don't run out of memory. Unbounded queues turn a slow consumer into a process-wide crash, which loses everything rather than something.

You cannot have all three. The overflow policy is where a broker declares which one it gives up, and the point of this toy is that the two policies that look like they're trading off how much you lose are actually trading off something else entirely.

3. Background you need

The two starred rows carry the result: once the queue is saturated, loss is arrival − service integrated over time, and capacity only shifts when saturation starts.

4. The mental model

Think of the queue as a bucket with a fixed volume, a tap filling it faster than the drain empties it. Once it is full, every drop that goes in means a drop that goes out — the only choice is which drop.

The argument in one sentence: both policies leave the queue at exactly capacity after an overflowing arrival, so the occupancy trajectory is policy-independent, so the count of displaced messages is policy-independent. Which message got displaced is the only degree of freedom left.

5. Reading the source

Here is the whole decision, and the docstring states the result before the code does:

    def offer(self, seq: int, published_at: int) -> bool:
        """Try to enqueue one message. Returns whether it was accepted.

        Both drop policies leave the queue at exactly `capacity` afterwards.
        That is not a coincidence and it is the whole result: the occupancy
        trajectory does not depend on which message you throw away, so the
        *count* of thrown-away messages cannot depend on it either.
        """
        if not self.full:
            self.queue.append((seq, published_at))
            return True
        if self.policy == DROP_NEWEST:
            self.dropped += 1
            return False
        if self.policy == DROP_OLDEST:
            self.queue.popleft()
            self.queue.append((seq, published_at))
            self.dropped += 1
            return False
        raise AssertionError("a BLOCK subscriber is never offered a message "
                             "it has no room for; the broker checks first")

broker.py · lines 58–78

Look at the two policy branches side by side. DROP_NEWEST does nothing to the queue; DROP_OLDEST does a popleft and an append. Both increment dropped by exactly one, and both leave len(self.queue) at capacity. There is no path through this method where one policy enqueues and the other doesn't. That symmetry is §6's headline — the demo is a demonstration of something you can read off the control flow.

The raise on the last branch is a claim about the broker, not a defensive default: a blocking subscriber is never offered a message it has no room for, because publish checks first. test_broker.py::a_blocking_subscriber_is_never_offered_a_message_it_cannot_hold is what stops that from rotting into a lie.

Now the fan-out, where backpressure is decided:

        """Fan one message out to every subscriber on the topic.

        Backpressure is a property of the *topic*, not of one consumer: if a
        blocking subscriber has no room, nobody gets the message, because the
        alternative is to deliver it to some subscribers and not others and
        call that the same stream.
        """
        subscribers = self.topics.get(topic, [])
        if any(s.policy == BLOCK and s.full for s in subscribers):
            self.refused += 1
            return False
        seq = self.published
        self.published += 1
        for s in subscribers:
            s.offer(seq, now)
        return True

broker.py · lines 105–120

The any(...) is the line that turns one subscriber's problem into everyone's. Note where self.published += 1 sits: after the refusal check, so a refused publish never gets a sequence number. That's what makes published + refused == publish_rate * ticks an identity rather than an approximation, and it is why §6.4's 791 is comparable to the drop policies' 791 — both count messages that never reached the slow subscriber.

6. The demo, and what it proves

The scenario: one topic, a publisher emitting 5 messages per tick for 200 ticks, and two subscribers — sub0 draining 1/tick (the slow one) and sub1 draining 5/tick. Every queue holds 10.

6.1 The two policies lose the same amount

Derive the 791. The publisher emits 5 × 200 = 1000. The subscriber consumes 1 per tick for 200 ticks, so it can absorb at most 200 — and it delivers 199, because one tick's drain happens before the first publish, hitting an empty queue. The queue itself holds 10 at the end. So:

The rate deficit is 5 − 1 = 4 messages per tick, over 200 ticks: 800. The buffer absorbed 9 of those, once, while it was filling. 800 − 9 = 791.

6.2 What the policy actually changed

Same count, opposite ends of the stream. drop-newest delivers a contiguous prefix 0,1,2,…,7 and then falls behind, ending on message 945 while the publisher is on 999. drop-oldest skips immediately to every fifth message and stays current, ending on 985.

The staleness gap is the number that matters operationally: 9.7 ticks against 2.0, nearly 5×. It is bounded, not unbounded — a queue of 10 draining at 1/tick can hold a message for at most 10 ticks, which is why drop-newest's final age is exactly 10 and test_broker.py::drop_newest_runs_five_times_staler asserts max(ages) <= capacity / drain_rate.

That bound is the whole basis for choosing between them. If your messages are state updates — a temperature, a cursor position, a cache invalidation — drop-oldest gives a consumer that is at most capacity/drain_rate ticks behind reality. If they are an event log where ordering and prefix-continuity matter, drop-newest gives you a clean prefix and a known truncation point.

6.3 The loss is arithmetic, and the policy is not in it

The prediction is

exact at every row. test_broker.py::the_loss_formula_is_exact_wherever_a_tick_burst_fits checks it across 320 configurations of (rate, capacity, drain), and ::the_policies_agree_even_outside_the_formulas_domain checks the equality across a further 210 where the formula does not apply — 530 in total, zero disagreements between the two policies.

Read the capacity column again. Going from 10 slots to 500 — fifty times the memory — takes loss from 791 to 301. It bought exactly 490 messages, which is the increase in capacity, once, against a deficit that runs at 4/tick forever. The buffer is a one-time credit, not a rate. That is why "just make the queue bigger" fails as a fix for a consumer that is persistently too slow, and works fine for one that is briefly too slow.

The other two sweeps make the same point from the other side:

Both variables move loss by ~200 per unit — ticks, exactly as the formula says. The policy column never moves it at all.

6.4 Backpressure: the one policy that changes the number

Nothing is dropped — and 791 publishes were refused. The same 791. The deficit is conserved: whichever policy you pick, 791 messages do not reach the slow subscriber, and the policy only chooses whether they die at the queue or are never born at the publisher.

What backpressure actually changed is who else pays. sub1 was keeping up perfectly and receiving all 995 messages under either drop policy. Under block it receives 208. A subscriber that had no problem now gets 79% less traffic, because a different subscriber is slow.

That is the exact moment "one slow subscriber doesn't block the others" stops being true, and it is a policy choice rather than a property of pub/sub.

6.5 The boundary: where it all vanishes

The effect requires sustained overload. Publish at 1/tick into a subscriber draining 1/tick and both policies drop zero; the queue never saturates, offer never reaches a policy branch, and the choice is inert (test_broker.py::nothing_is_lost_when_the_subscriber_keeps_up).

The same is true for a burst that fits: at publish_rate=2, drain=1, capacity 201 absorbs the entire 200-tick run with zero loss, while capacity 199 loses 2. Below that knife-edge the buffer is doing its actual job — absorbing a transient mismatch — and above it the buffer is only delaying an arithmetic certainty. A reader deciding whether to enlarge a queue in production is really asking which side of that line their traffic is on.

7. Design decisions and roads not taken

Per-subscriber queues, not one shared queue. This is the design decision that makes the toy a pub/sub toy rather than a queue toy. With one shared queue, a slow consumer's backlog is everyone's backlog and §6.4 would be the only possible outcome. Per-subscriber queues are what make isolation the default and backpressure a choice.

Integer ticks, drained before published. No clock, no sleeping, no threads, no randomness — the numbers on this page are the same on every machine, which for a toy about counting lost messages is the difference between a result and an anecdote. Draining before publishing within a tick is a choice, and run's docstring claims it shifts every number by one queue slot without changing the result. That is verified rather than asserted: reversing the order gives 200 delivered instead of 199, and 791 dropped either way (cf_broker.py CF1).

block refuses the whole publish, not just the full subscriber. The alternative — deliver to whoever has room, skip the rest — is worse than it looks: it silently converts backpressure back into loss, and the subscribers that "succeeded" now have a different message stream from the ones that didn't. Making it topic-wide keeps the semantics honest, at the cost of §6.4.

No drop-random, no priority, no TTL. They're all real policies — TTL in particular is what most brokers actually reach for — but each is a different axis, and the point here is that two policies which look like they differ on volume differ only on identity. A third policy that also doesn't change the volume would add length without adding an idea.

Loss is counted, latency is not. There is an ages() method and §6.2 uses it, but the toy doesn't model per-message processing cost or queueing delay in any realistic way. A toy that took latency seriously would need a service time distribution and would be a queueing theory toy instead.

8. What's simplified vs. the real thing

• ZeroMQ makes this exact choice, and encodes it in the socket type. A PUB socket that reaches its high-water mark for a subscriber drops — "any messages that would be sent to the subscriber in question shall instead be dropped until the mute state ends" — and note it is per subscriber, which is §7's first decision. A PUSH socket blocks: "any send operations on the socket will block until the mute state ends." Same library, opposite policies, chosen by what the socket is for.
• Kafka doesn't drop, it blocks then throws. The producer buffers into buffer.memory and, when full, blocks for up to max.block.ms before raising. That's the block policy with a timeout — and the timeout is itself a third answer, converting backpressure into a publisher-side error rather than either loss or an indefinite stall.
• NATS drops the subscriber, not the message. A client that falls too far behind is cut off with a "slow consumer" error, on the theory that a consumer which cannot keep up is better off knowing than being quietly starved of messages it will never catch up on.
• Real brokers persist. Kafka's log lets a slow consumer fall arbitrarily behind at the cost of disk, converting a memory bound into a retention window. That changes the constant in §6.3 dramatically but not the structure: retention is still a one-time credit measured in bytes or days, and a consumer slower than the producer still eventually loses.
• No acknowledgements, no redelivery, no consumer groups. delivered here means "removed from the queue", not "successfully processed". Every real broker distinguishes those, and the gap between them is where at-least-once delivery lives.
• The slow subscriber is slow at a constant rate. Real consumers are slow in bursts — a GC pause, a slow downstream call — which is exactly the transient case §6.5 identifies as the one where buffering genuinely works.
• One process, one thread, no network. There is no serialization cost, no partial delivery, and no possibility of the broker itself being the bottleneck.

9. Check yourself

Q1. The demo publishes 1000 messages and the slow subscriber receives 199, loses 791. That leaves 10 unaccounted for. Where are they?

Q2. Your subscriber drains 1/tick, the publisher sends 5/tick, and you are losing 79% of messages. Someone proposes increasing the queue from 10 to 1000. How many messages does that save over a 200-tick run, and what is the loss rate afterwards?

Q3. Under drop-oldest, the delivered sequence starts [0, 1, 5, 10, 15, 20, …]. Why does it jump by exactly 5 after the first two?

Q4. You have a topic with one block subscriber and nine drop-newest subscribers, all healthy except the blocking one. What happens to the other nine, and which line of broker.py decides it?

Q5. A colleague argues that drop-oldest must lose more, because it throws away a message it had already accepted and then accepts a new one — two messages touched instead of one. Where is the error?

10. Further reading

• ZeroMQ: Socket API — read the "mute state" sections for PUB and PUSH back to back. It is this page's §6 as a library design decision: the same high-water mark, drop for publishers and block for pipelines, with the reasoning stated for each socket type.
• Kafka producer configuration — buffer.memory and max.block.ms. The producer-side buffer is §6.3's capacity term, and max.block.ms is what turns an indefinite stall into a bounded one.
• NATS: slow consumers — the fourth policy this toy doesn't implement: disconnect the subscriber. Worth reading for the argument about why silent dropping is considered the worse failure.
• Little's law — the queueing-theory statement of §6.3's arithmetic. L = λW: with L bounded by capacity and λ fixed by the publisher, the only free variable is how long a message waits, which is exactly the age measurement in §6.2.
• Welsh & Culler, Adaptive Overload Control for Busy Internet Servers — the SEDA follow-up, on what a server should do when it cannot keep up. Makes the case that the interesting knob is admission control at the front door rather than eviction policy at the queue, which is §6.4's conclusion arrived at from the other direction.