Pub/sub fan-out

Deep dive

How it works

Parts:

• subscription index — exact topics in a hash map, wildcards in a topic-matching trie
• shared frame — the message encoded once, held by Arc
• connection queue — bounded byte budget per subscriber, holds frame references
• overflow policy — drop oldest, drop newest or disconnect
• connection writer — drains a queue with one writev per wakeup

Operations: subscribe, publish, overflow, drain (steps below).

1. Subscribe: record (topic or pattern) -> connection. Exact topics sit in a hash map; wildcards need a topic-matching trie.
2. Publish: look up the matching connections. Deduplicate: a connection matching by two subscriptions gets the message once (Redis, by contrast, delivers once per matching subscription).
3. Serialize the frame once. Push a reference (Arc) to it onto each connection's queue. Never copy the payload per subscriber.
4. Check the queue's byte budget before pushing. On overflow apply the policy: drop oldest (state-like feeds where only the latest matters), drop newest, or disconnect (the client reconnects and resyncs from a snapshot, ideally via sequence numbers).
5. Per-connection writer: drain up to a byte budget per wakeup and send it with one writev/send, so syscalls scale with wakeups, not messages.

publisher --1 frame--> [broker]--Arc-->[q: c1] --writev--> socket c1
                          |------Arc-->[q: c2] --writev--> socket c2 (slow: q full -> policy)
                          '------Arc-->[q: c3] --writev--> socket c3

Code

Single-threaded core: exact topics, shared payloads, byte-bounded queues, two slow-consumer policies, batched drain. Compiled and tested (asserts on delivery counts, drop and disconnect behaviour, Arc::ptr_eq sharing).

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;

pub type SubId = u64;
pub type Msg = Arc<[u8]>; // one allocation per publish, N refcount bumps

#[derive(Clone, Copy, PartialEq, Debug)]
pub enum SlowPolicy {
    DropOldest, // lossy feeds: keep the newest (quotes, cursors)
    Disconnect, // like Redis/NATS: a client that cannot keep up is cut off
}

pub struct Subscriber {
    q: VecDeque<Msg>,
    bytes: usize,
    max_bytes: usize, // bound by bytes, not message count
    policy: SlowPolicy,
    pub dropped: u64,
}

#[derive(Default)]
pub struct Broker {
    topics: HashMap<String, Vec<SubId>>,
    subs: HashMap<SubId, Subscriber>,
    next_id: SubId,
}

impl Broker {
    pub fn connect(&mut self, max_bytes: usize, policy: SlowPolicy) -> SubId {
        self.next_id += 1;
        self.subs.insert(
            self.next_id,
            Subscriber { q: VecDeque::new(), bytes: 0, max_bytes, policy, dropped: 0 },
        );
        self.next_id
    }

    pub fn subscribe(&mut self, id: SubId, topic: &str) {
        let v = self.topics.entry(topic.to_string()).or_default();
        if !v.contains(&id) {
            v.push(id);
        }
    }

    pub fn disconnect(&mut self, id: SubId) {
        self.subs.remove(&id);
        for v in self.topics.values_mut() {
            v.retain(|&s| s != id);
        }
        self.topics.retain(|_, v| !v.is_empty());
    }

    /// Returns the number of subscribers the message was queued for.
    pub fn publish(&mut self, topic: &str, payload: &[u8]) -> usize {
        let Some(ids) = self.topics.get(topic) else { return 0 };
        let msg: Msg = Arc::from(payload);
        let mut delivered = 0;
        let mut cut = Vec::new();
        for id in ids {
            let s = self.subs.get_mut(id).expect("topic index and subs agree");
            while s.bytes + msg.len() > s.max_bytes {
                match s.policy {
                    SlowPolicy::DropOldest => match s.q.pop_front() {
                        Some(old) => { s.bytes -= old.len(); s.dropped += 1; }
                        None => break, // message alone exceeds the cap
                    },
                    SlowPolicy::Disconnect => break,
                }
            }
            if s.bytes + msg.len() > s.max_bytes {
                s.dropped += 1;
                if s.policy == SlowPolicy::Disconnect { cut.push(*id); }
                continue;
            }
            s.bytes += msg.len();
            s.q.push_back(Arc::clone(&msg));
            delivered += 1;
        }
        for id in cut { self.disconnect(id); }
        delivered
    }

    /// Hand one connection up to `budget` bytes in a single batch (one writev).
    pub fn drain(&mut self, id: SubId, budget: usize) -> Vec<Msg> {
        let mut out = Vec::new();
        let Some(s) = self.subs.get_mut(&id) else { return out };
        let mut used = 0;
        while let Some(m) = s.q.front() {
            if !out.is_empty() && used + m.len() > budget { break; }
            used += m.len();
            s.bytes -= m.len();
            out.push(s.q.pop_front().unwrap());
        }
        out
    }

    pub fn is_connected(&self, id: SubId) -> bool { self.subs.contains_key(&id) }
}

Run it behind one lock per shard, or give each shard thread its own Broker and route by hash(topic); keep the socket writes outside the lock.

Complexity and constants

• Publish: O(N) pushes, each an Arc clone (one atomic increment, order of a few ns uncontended, much more when many threads share the same payload) and a 16-byte fat pointer in the queue.
• Copying instead would cost N x payload: 10,000 subscribers x 1 KB = 10 MB of memcpy per message.
• Memory worst case: N x max_bytes. 100,000 connections at 64 KiB is 6.4 GB: size the cap from the budget, not the other way round.
• The real cost at scale is syscalls: one send per connection per wakeup (order of 1 us each plus TLS/framing work), and cache misses chasing N queues.

Efficiency

• Cost: publish is O(N) pushes, each an Arc clone (one atomic increment, order of a few ns uncontended) plus a 16-byte fat pointer; the payload is never copied. Copying instead costs N x payload: 10,000 subscribers x 1 KB = 10 MB per message. Memory worst case N x max_bytes (100,000 connections at 64 KiB = 6.4 GB). At scale the cost is one send per connection per wakeup (order of 1 µs plus TLS/framing). Bound: CPU in per-connection send work, then memory bandwidth, network last.
• Levers:
  • encode the wire frame once — O(1) CPU per publish instead of O(N) — always
  • batch per connection (one writev per wakeup) — fewer syscalls and packets, latency bounded by the wakeup interval — busy topics
  • conflate by key — a slow reader costs O(keys), not O(rate x time) — state-like feeds
  • shard by topic hash (Phoenix.PubSub: one partition per 4 cores by default; Redis 7.0 sharded pub/sub) — each core touches its own lists — many topics, many cores
  • in-process tokio::sync::broadcast — each value stored once, cloned per receiver on demand — few local subscribers
• Naive → standard → state of the art:
  • encode and send per subscriber · O(N) CPU and N x payload copies
  • shared Arc frame, bounded queues · O(N) pointer pushes, 0 payload copies
  • plus batched writev, conflation and sharding · syscalls scale with wakeups, not messages

Implementation notes

• Bound by bytes, not message count; message sizes vary by orders of magnitude.
• Soft + hard limits (Redis style) tolerate short bursts and still catch a stuck reader: the hard limit is an absolute cap, the soft limit counts only if exceeded continuously for N seconds.
• Keep topic lookup and queue push under the lock; do socket I/O in a per-connection task that only drains. Never write to a socket while holding the subscriber-table lock.
• Detect dead peers separately (heartbeats / TCP keepalive); a vanished client otherwise just looks like a slow one until its queue fills.
• On slow consumers, count and expose dropped per connection so operators see the loss.
• Cross-node fan-out: send one copy per remote node (not per remote subscriber) and let that node fan out locally. NATS routes by interest the same way.

Pitfalls

• Unbounded per-subscriber queues: one stalled client turns into an out-of-memory kill for everyone.
• Bounding by message count with variable-size payloads.
• Holding the broker lock during write: one slow socket blocks all publishers.
• Per-subscriber serialization or per-subscriber payload clones.
• Dropping silently: with no gap signal (sequence number), the client can't know it missed data. Disconnect-and-resync is often safer than quiet drops.
• Expecting delivery guarantees: publishes to a topic with no subscriber vanish; reconnecting clients miss everything in between.
• Assuming ordering across topics or across publishers; only per-publisher, per-connection FIFO holds.

Variants and alternatives

In production

Redis Pub/Sub (channels, patterns, sharded since 7.0), NATS core subjects with wildcards and slow-consumer cut-off, Phoenix.PubSub behind Phoenix Channels, MQTT brokers for device telemetry (topic filters, QoS 0 is the same at-most-once model), WebSocket gateways for chat and live data. Redis client-side caching can deliver invalidation messages over Pub/Sub (RESP2 redirect mode).

Implementations

Sources

• Patrick Th. Eugster, Pascal A. Felber, Rachid Guerraoui, Anne-Marie Kermarrec, The Many Faces of Publish/Subscribe, ACM Computing Surveys 35(2), 2003. paper
• Redis documentation, Pub/Sub. docs; PUBLISH complexity: docs; buffer limits: redis.conf
• NATS documentation, Core NATS: publish-subscribe. docs; MAX_PENDING_SIZE: server/const.go
• Tokio, tokio::sync::broadcast. docs
• Phoenix.PubSub. docs
• OASIS, MQTT Version 3.1.1 (publish/subscribe, QoS 0). spec