Market Data Fan-Out

Two designs share the same skeleton (a ring of slots indexed by a monotonically increasing sequence number) and differ in one rule:

• Gating (Disruptor). The producer tracks every consumer's sequence and won't overwrite a slot until all of them have passed it. Nothing is lost, and the slowest consumer sets everyone's pace. Use it for internal consumers that must not miss anything and that you control: journal, replication, risk.
• Overwriting (broadcast). The producer never looks at consumers and overwrites the oldest slot. Each consumer detects for itself that its next slot was overwritten and reports how much it lost. The publisher's cost is independent of the number and speed of readers. Use it for market data to anyone you don't control.

The ring has a capacity that is a power of two, so seq & mask is the slot index. The writer owns next_seq. Each reader owns a cursor, next, the sequence it wants to read next. Every slot stores the sequence number of the value it currently holds. A reader looking at slot next & mask sees one of three things:

• slot.seq == next: the value it wants. Take it and advance.
• slot.seq < next, or never written: not published yet, so the reader is caught up.
• slot.seq > next: the writer has lapped this reader and overwritten its slot. It lost data. Resync to the oldest value still in the ring (next_seq − capacity) and report how many it skipped.

You'll write this single-threaded first so the semantics are exact and testable. Day 3 adds the atomics. The per-slot stamp is what makes the concurrent version possible: a reader decides everything from the one slot it's looking at, without a consistent view of the writer's cursor.

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Recv<T> {
    Item(T),
    Empty,
    Lagged(u64),
}

/// Each slot carries the sequence number of the value it holds.
#[derive(Clone, Copy)]
struct Slot<T> {
    seq: u64, // u64::MAX = never written
    value: T,
}

pub struct BroadcastRing<T: Copy + Default> {
    slots: Vec<Slot<T>>,
    mask: u64,
    next_seq: u64, // seq the next publish gets; the first publish is seq 0
}

pub struct Reader {
    next: u64, // next seq this reader wants
}

impl<T: Copy + Default> BroadcastRing<T> {
    pub fn new(capacity: usize) -> Self {
        assert!(capacity.is_power_of_two(), "capacity must be a power of two");
        BroadcastRing {
            slots: vec![Slot { seq: u64::MAX, value: T::default() }; capacity],
            mask: capacity as u64 - 1,
            next_seq: 0,
        }
    }

    /// Never blocks and never looks at readers. Returns the assigned seq.
    pub fn publish(&mut self, value: T) -> u64 {
        todo!()
    }

    /// A new reader starts at the live edge: it sees only later publishes.
    pub fn subscribe(&self) -> Reader {
        todo!()
    }

    /// Item(v) if the reader's slot holds seq == r.next (then advance);
    /// Empty if that seq isn't published yet;
    /// Lagged(n) if the slot was overwritten: jump r.next to the oldest seq
    /// still in the ring (next_seq - capacity) and report n skipped.
    pub fn read(&self, r: &mut Reader) -> Recv<T> {
        todo!()
    }
}

With the writer and readers on different threads, a reader can be copying a slot at the moment the writer overwrites it. The standard fix is a seqlock per slot. The writer marks the slot as being written, writes the payload, then publishes the new sequence. The reader reads the stamp, copies the payload, then re-reads the stamp. If the stamp changed, the copy may be torn and is discarded.

// Writer (one thread). Payload words are AtomicU64 so concurrent reads are not UB.
slot.seq.store(WRITING, Ordering::Relaxed);
std::sync::atomic::fence(Ordering::Release);      // the marker becomes visible no later than the payload
for (w, v) in slot.words.iter().zip(payload) { w.store(v, Ordering::Relaxed); }
slot.seq.store(seq, Ordering::Release);           // publish: payload happens-before this

// Reader (many threads).
let s1 = slot.seq.load(Ordering::Acquire);
if s1 != want { /* Empty or Lagged, decided as in the drill */ }
let copy = slot.words.each_ref().map(|w| w.load(Ordering::Relaxed));
std::sync::atomic::fence(Ordering::Acquire);      // the payload loads complete before the re-check
let s2 = slot.seq.load(Ordering::Relaxed);
if s2 != s1 { /* overwritten while copying: treat as Lagged */ }

Two subtleties catch everyone. First, the payload must be atomic (or copied with operations the memory model allows to race). In Rust, as in C++, a plain read racing with a write is undefined behaviour even if you then discard the value, and "it's just a memcpy" isn't a defence. For small fixed-size updates, a few AtomicU64 words loaded with Relaxed compile to ordinary loads on x86-64. Second, the fences are not optional. Without the reader's Acquire fence, the second stamp load can be satisfied before the payload loads, and the check proves nothing.

Reader cursors belong to their readers. The writer of an overwriting ring never reads them. But if 16 cursors sit in one Vec<u64>, they share two cache lines, and every reader's increment invalidates its neighbours' copies. Pad each cursor to its own line (CachePadded from week 4). The slots are the opposite case: every reader reads the line the writer just wrote, so each publish sends that line to 16 cores. You can't avoid that, but you can amortize it. Pack several small updates per cache line and have readers consume runs of slots per wakeup.

When a consumer falls behind, there are only three honest options:

1. Drop and recover. The consumer notices a sequence gap (its Lagged), discards its book, and rebuilds from a snapshot plus later incrementals, exactly as in week 10. This is correct for any consumer, but recovery takes time and a snapshot.
2. Conflate. Instead of queueing every update for a slow consumer, keep only the latest value per key (per price level, for L2). The consumer receives fewer messages, and after it drains it holds the same state it would have had from every update. This is ideal for book feeds to humans and slow systems.
3. Disconnect. For consumers that must see every event and can't keep up, cut them off and let them reconnect and recover. It's honest and simple, and it protects everyone else.

The details of conflation decide whether it's correct. Keep pending levels in the order they first became dirty, so a level updated constantly doesn't starve others. Overwrite a pending entry in place. Deliver deletes (qty 0) like any other value. Re-queue a level at the back if it changes again after being popped. Memory is bounded by the number of distinct levels, not the number of updates.

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

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Side {
    Bid,
    Ask,
}

/// L2 update: the level's new total quantity (0 = level removed).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct LevelUpdate {
    pub side: Side,
    pub price: i64,
    pub qty: u64,
    pub seq: u64,
}

pub struct Conflator {
    // Your fields. One option: a map from (side, price) to the latest
    // pending update, plus a FIFO of keys in first-dirty order.
}

impl Conflator {
    pub fn new() -> Self {
        todo!()
    }

    /// Record an update. If its level is already pending, replace the
    /// pending value in place (its queue position doesn't change).
    pub fn push(&mut self, u: LevelUpdate) {
        todo!()
    }

    /// The pending level that became dirty earliest, carrying its latest
    /// qty and seq. Once popped, a level that changes again goes to the back.
    pub fn pop(&mut self) -> Option<LevelUpdate> {
        todo!()
    }

    /// Number of distinct pending levels.
    pub fn len(&self) -> usize {
        todo!()
    }
}

A WebSocket gateway is a reader of the broadcast ring that serves many clients:

• One ring reader per gateway thread, not per client. The gateway reads each update once, encodes it once into a shared buffer (binary frames, or JSON if you must), and copies the bytes into each subscribed client's outbound queue.
• Per-client bounded queues with a policy. Book channels conflate per level when the queue is full. Trade channels disconnect. TCP's send buffer is a second queue you don't control, so watch its fill level through the writer's progress, not just your own queue.
• Snapshot on subscribe. A new subscriber gets the book at sequence N, then only incrementals with seq above N, the week 10 protocol over a new transport.
• Compression is a trade-off. permessage-deflate (RFC 7692) cuts bandwidth for text feeds but costs CPU per client per message, and its compression context is per connection, so you can't encode once and send to many. Most latency-sensitive feeds skip it.

Inside a colocation facility, exchanges publish market data over UDP multicast. The publisher sends each packet once, and switches replicate it to every subscribed port. The fan-out cost moves into hardware, the publisher's cost doesn't depend on subscriber count, and there's no slow-consumer problem at the source: a slow subscriber drops packets in its own NIC or socket buffer and has to recover by itself.

UDP doesn't retransmit, so feeds are built to be recoverable. Each message carries a sequence number (MoldUDP64, MDP 3.0 packet headers). The same stream is published twice on separate networks (A/B feeds), and receivers take whichever copy of each sequence number arrives first. Gaps that both lines miss are filled from a snapshot or recovery channel, or a request-response retransmission service. This is your broadcast ring's Lagged → snapshot path, distributed across a network.