Lab IV: A Mini Market — Venue + Broker End-to-End
Before you start. This lab assembles almost everything from Part V: ch23 (venue architecture: gateway → sequencer → engine), ch25 (feed publishing, snapshots, gap recovery), and ch26 (broker adapters, SOR, the double-fill race) — plus ch13 (event sourcing and replayability) and the mental model from ch00f.
You have built a matching engine before. You have built a market-data pipeline before. What you have probably never done is run both sides of the market at once and watch them argue with each other: the venue dropping deltas on a slow subscriber, the broker detecting the gap and routing around the blind spot, the SOR refusing to re-route until a cancel-ack comes back. The interesting interview questions come out of that argument, and after this lab you will have watched it happen in your own terminal.
One word before anything else: a delta is one incremental book change — “level 10004 now has 250” — the small message a venue publishes instead of resending the whole book every time anything moves. Drop one and your copy of the book silently diverges from the venue’s; most of this lab’s drama comes from exactly that.
Everything runs in one process on your Mac: threads + crossbeam-channel, no tokio, no sockets. Every channel boundary is labeled with what it would be in production (TCP session, multicast group), so the logic transfers 1:1.
VENUE A (thread) VENUE B (thread)
gateway→sequencer→engine gateway→sequencer→engine
│ │ │ │
acks/fills feed (L2 incr+snap) acks/fills feed
└───┬────┴───────┬────────────────┬────────┘
▼ ▼ ▼
[broker: venue adapters × 2 → normalized books]
│
[SOR: cost model picks venue(s), splits]
│
[parent-order manager: fills, re-routes, client report]
▲
client script: sends parent orders
(In the diagram: L2 = price-level depth — the whole ladder of prices and sizes, not just the best bid/ask; incr+snap = incremental deltas plus a periodic snapshot, so a late or gapped subscriber can always rebuild.)
Venue A is configured fast but wide (2 ms, deep book, 0.6-tick fee), Venue B slow but tight (20 ms, thin book, 0.2-tick fee) — so routing is a genuine trade-off, not a foregone conclusion. Unpacking the trader shorthand: wide = worse prices at the top of the book, deep = lots of size resting behind them; tight = better prices at the top, thin = less size behind them — so which venue is “cheapest” depends on how much you’re buying.
Step 0 — Workspace
mini-market/
├── Cargo.toml
└── src/
├── main.rs # scenarios + determinism check
├── types.rs # shared message types
├── book.rs # price-time matching engine
├── venue.rs # gateway → sequencer → engine → feed
├── adapter.rs # broker-side feed handler (normalized book)
└── sor.rs # cost model, splitter, parent-order manager
# Cargo.toml
[package]
name = "mini-market"
version = "0.1.0"
edition = "2021"
[dependencies]
crossbeam-channel = "0.5"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
Step 1 — Shared types
Fixed structs everywhere on the hot path; serde only for the human-readable client report at the end. One convention to notice before you scroll: Px is an integer count of ticks — floats never touch a price (ch00d’s tick-size rule) — so $100.00 is 10_000 everywhere in this lab.
#![allow(unused)]
fn main() {
// src/types.rs
use serde::Serialize;
pub type Px = i64; // integer ticks (1 tick = $0.01); 10_000 = $100.00
pub type Qty = u64;
pub type Seq = u64;
// #[derive(...)] is like a decorator that writes boilerplate at compile time:
// Clone/Copy/Debug for copying and printing, Serialize (serde) for JSON later.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub enum Side { Buy, Sell }
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize)]
pub enum VenueId { A, B }
#[derive(Clone, Debug)]
pub struct NewOrder { pub client_id: String, pub side: Side, pub px: Px, pub qty: Qty }
/// Order entry. In production: FIX/OUCH over a per-member TCP session.
#[derive(Clone, Debug)]
pub enum GatewayIn { New(NewOrder), Cancel { client_id: String }, CancelAll }
#[derive(Clone, Debug, PartialEq, Serialize)]
pub enum ExecType { Ack, Reject(String), Fill { px: Px, qty: Qty }, CancelAck { qty: Qty } }
/// Private execution report back to the order sender.
#[derive(Clone, Debug, Serialize)]
pub struct ExecReport { pub venue: VenueId, pub client_id: String, pub exec: ExecType, pub seq: Seq }
#[derive(Clone, Debug, Serialize)]
pub struct Level { pub px: Px, pub qty: Qty }
/// Public market data. In production: seq-numbered UDP multicast per channel.
#[derive(Clone, Debug)]
pub enum FeedMsg {
Delta { seq: Seq, side: Side, px: Px, qty: Qty }, // qty = new total at level; 0 = level removed
Snapshot { seq: Seq, bids: Vec<Level>, asks: Vec<Level> },
}
#[derive(Serialize)]
pub struct FillLine { pub venue: VenueId, pub px: Px, pub qty: Qty }
#[derive(Serialize)]
pub struct ClientReport { pub parent: String, pub filled: Qty, pub avg_px: f64, pub fills: Vec<FillLine> }
}Step 2 — The matching engine
Minimal price-time book — you have built a bigger one; this one exists to emit the right events, not to be fast. The two data structures below carry the two priorities directly: BTreeMap keeps the price levels permanently sorted (price priority), and each level’s VecDeque is a first-come-first-served queue of resting orders (time priority).
“The right events” is MatchOut, the triple every submit returns: fills for the taker (the incoming order), fills for the makers it hit (the resting orders on the other side), and the book deltas the feed will publish. That triple is exactly what the venue thread in Step 3 fans out — private reports to each party, public deltas to everyone.
#![allow(unused)]
fn main() {
// src/book.rs
use crate::types::*;
use std::collections::{BTreeMap, VecDeque};
// One price level = a first-come-first-served queue: first to arrive is filled
// first — time priority. (VecDeque = a double-ended queue;
// each entry is one resting order, (client_order_id, qty).)
type LevelQ = VecDeque<(String, Qty)>;
// The two sides. A BTreeMap stores its keys permanently sorted, so "best
// price" is just the first (or last) key — no searching, ever.
#[derive(Default)]
pub struct Book { bids: BTreeMap<Px, LevelQ>, asks: BTreeMap<Px, LevelQ> }
pub struct MatchOut {
pub taker_fills: Vec<(Px, Qty)>,
pub maker_fills: Vec<(String, Px, Qty)>,
pub deltas: Vec<(Side, Px, Qty)>, // (book side, price, new aggregate qty)
}
// Total quantity resting at one price: sum every order in that level's queue.
// (Each entry is a tuple (client_order_id, qty); `.0`/`.1` pick tuple fields by
// position, so `e.1` is the qty. You'll see this indexing throughout the file.)
fn lvl_qty(q: &LevelQ) -> Qty { q.iter().map(|e| e.1).sum() }
impl Book {
/// One side's full ladder, best price first — the snapshot payload.
pub fn depth(&self, side: Side) -> Vec<Level> {
match side {
// maps iterate ascending, so bids are reversed: highest (best) bid first
Side::Buy => self.bids.iter().rev().map(|(p, q)| Level { px: *p, qty: lvl_qty(q) }).collect(),
Side::Sell => self.asks.iter().map(|(p, q)| Level { px: *p, qty: lvl_qty(q) }).collect(),
}
}
/// Match an incoming order against the opposite side; any remainder rests.
/// Returns the MatchOut triple: taker fills, maker fills, feed deltas.
pub fn submit(&mut self, o: &NewOrder) -> MatchOut {
let (mut tf, mut mf, mut deltas, mut qty) = (Vec::new(), Vec::new(), Vec::new(), o.qty);
loop {
let opp = match o.side { Side::Buy => &mut self.asks, Side::Sell => &mut self.bids };
// best opposing price: lowest ask (first key) when buying, highest bid (last key) when selling
let best = match o.side { Side::Buy => opp.keys().next().copied(), Side::Sell => opp.keys().next_back().copied() };
let Some(px) = best else { break }; // nobody left on the other side of the market — stop. (let-else: unpack the Some or bail)
let crosses = match o.side { Side::Buy => px <= o.px, Side::Sell => px >= o.px };
if !crosses || qty == 0 { break; }
let q = opp.get_mut(&px).unwrap();
while qty > 0 {
let Some(front) = q.front_mut() else { break };
let take = qty.min(front.1);
mf.push((front.0.clone(), px, take));
tf.push((px, take));
front.1 -= take; qty -= take;
if front.1 == 0 { q.pop_front(); }
}
let left = lvl_qty(q);
if left == 0 { opp.remove(&px); }
deltas.push((if o.side == Side::Buy { Side::Sell } else { Side::Buy }, px, left));
}
if qty > 0 { // remainder rests at its limit, at the back of the level queue
let map = match o.side { Side::Buy => &mut self.bids, Side::Sell => &mut self.asks };
// entry(): fetch that price's level, creating it if it doesn't
// exist yet; push_back = join the back of the queue (time priority kept)
map.entry(o.px).or_default().push_back((o.client_id.clone(), qty));
deltas.push((o.side, o.px, lvl_qty(&map[&o.px])));
}
MatchOut { taker_fills: tf, maker_fills: mf, deltas }
}
/// Remove one client's resting order, wherever it sits on the book.
pub fn cancel(&mut self, cid: &str) -> Option<(Side, Px, Qty, Qty)> { // (side, px, canceled, level left)
for (side, map) in [(Side::Buy, &mut self.bids), (Side::Sell, &mut self.asks)] {
let hit = map.iter().find(|(_, q)| q.iter().any(|e| e.0 == cid)).map(|(p, _)| *p);
if let Some(px) = hit {
let q = map.get_mut(&px).unwrap();
let before = lvl_qty(q);
q.retain(|e| e.0 != cid); // pull this one client out of the line; everyone else keeps their place (retain = filter in place)
let left = lvl_qty(q);
if left == 0 { map.remove(&px); }
return Some((side, px, before - left, left));
}
}
None
}
/// Kill-switch mass cancel: clear the whole book, sparing "SEED" background liquidity.
pub fn cancel_all(&mut self) -> Vec<(String, Qty)> { // returns canceled non-seed orders
let mut out = Vec::new();
for map in [&mut self.bids, &mut self.asks] {
for q in map.values() { for e in q { if e.0 != "SEED" { out.push((e.0.clone(), e.1)); } } }
map.clear();
}
out
}
}
}Step 3 — The venue: gateway → sequencer → engine → two outputs
One thread per venue. The gateway checks run before the sequencer, so a rejected order never consumes a sequence number — the sequence is the replayable truth of the market, and a reject never happened as far as the market is concerned. The sequencer is simply the fact that one consumer drains the channel — consuming in arrival order is the total order (ch23). Two output paths: private exec reports to the sender, and a public seq-numbered feed with a periodic snapshot — the late-joiner contract from ch25: a new subscriber can only bootstrap from a snapshot, so one is published every snap_every deltas.
Two more things to watch for in the code. First, the seed_liq orders in the config: they stand in for other market participants’ resting quotes — background liquidity that isn’t yours — which is why Step 2’s cancel_all (the kill-switch path) deliberately spares anything tagged "SEED". Second, the two sequence-number spaces, seq and fseq — two independent counters, one numbering the private exec-report stream, one the public feed. The diagram below marks where each advances; they never sync and never need to — real venues keep them separate too.
This file is also where the lab’s Rust concurrency toolkit first appears. Node gives you one event loop — nothing runs at the same time as your code. This lab runs several such loops at once: each venue is its own OS thread, its for msg in rx.iter() loop genuinely executing in parallel with yours. Threads share no variables by default; the only doors between them are channels — typed one-way mail chutes, postMessage between workers, except receiving sleeps until a message drops in. Every ══ channel ══ line in the diagrams from here on is one such chute (a crossbeam-channel), labeled with the production wire it stands in for.
VENUE THREAD (one lane — one consumer draining order_rx IS the sequencer)
══ order_rx channel (thread boundary — prod: TCP order-entry session, FIX/OUCH) ══
(1) rx.iter() — order arrives
(2) GATEWAY: dup client_id? size ≤ max? collar ±10%? tokens left?
│ fail ─▶ exec_tx.send(Reject) — exits HERE, pre-sequencer: NO seq consumed
▼ pass
(3) SEQUENCER: seq += 1 ← private counter `seq`: numbers YOUR events
(4) ENGINE: book.submit(&o) → { taker fills, maker fills, deltas }
├─▶ (5) PRIVATE: exec_tx.send(Ack/Fill/CancelAck) — one `seq` each
│ ══ channel (prod: TCP session) ══ reliable: send() waits, never drops
└─▶ (6) PUBLIC: feed try_send(Delta/Snapshot) — one `fseq` each
══ channel, bounded (prod: UDP multicast) ══ best-effort: full slot =
DROPPED, never blocks. fseq contiguous: see 5 then 7 ⇒ 6 went missing
Each idiom gets a comment at first use, so read the comments as part of the text.
#![allow(unused)]
fn main() {
// src/venue.rs
use crate::{book::Book, types::*};
use crossbeam_channel::{unbounded, Receiver, Sender};
use std::collections::HashSet;
use std::thread::{self, JoinHandle};
use std::time::Duration;
pub struct VenueConfig {
pub id: VenueId, pub ref_px: Px, pub fee_ticks: f64, pub latency: Duration,
pub max_qty: Qty, pub gw_tokens: u32, pub snap_every: u64,
pub seed_liq: Vec<(Side, Px, Qty)>,
}
/// What callers hold: the order-entry sender, plus a JoinHandle — a handle to
/// the thread's eventual return value (its event log). Where JS would `await`
/// a Promise, a JoinHandle is redeemed with a blocking `.join()`.
pub struct VenueHandle { pub tx: Sender<GatewayIn>, pub join: JoinHandle<Vec<String>> }
/// Start the venue thread; return its order-entry sender and join handle.
/// In production `tx` is a TCP order-entry session and `feeds` are multicast groups.
pub fn spawn(cfg: VenueConfig, exec_tx: Sender<ExecReport>, feeds: Vec<Sender<FeedMsg>>) -> VenueHandle {
let (tx, rx) = unbounded(); // a mail chute with no depth limit: send always succeeds instantly, the pile just grows
// A JS closure *shares* what it captures; `move` instead hands the closure sole
// ownership of cfg, rx, exec_tx, feeds — it must, because the new thread outlives
// this function, and two threads may never share unguarded data.
let join = thread::spawn(move || run(cfg, rx, exec_tx, feeds));
VenueHandle { tx, join }
}
/// Fan one message out to every feed subscriber — multicast in miniature.
fn publish(feeds: &[Sender<FeedMsg>], msg: FeedMsg) {
// Each subscriber's feed queue is a mail slot of fixed depth. try_send delivers
// only if there's room — a stuffed slot means the letter is DROPPED and the venue
// moves on: a slow reader must never set the venue's pace (real UDP multicast
// behaves exactly like this when a subscriber lags).
for f in feeds { let _ = f.try_send(msg.clone()); } // let _ = "it can fail; dropping is the plan"
}
// Package the current book as a Snapshot feed message.
fn snap(book: &Book, seq: Seq) -> FeedMsg {
FeedMsg::Snapshot { seq, bids: book.depth(Side::Buy), asks: book.depth(Side::Sell) }
}
// Publish one delta, plus a fresh snapshot every `every` deltas (the late-joiner contract).
fn delta(book: &Book, feeds: &[Sender<FeedMsg>], fseq: &mut Seq, since: &mut u64, every: u64,
side: Side, px: Px, qty: Qty) {
*fseq += 1; publish(feeds, FeedMsg::Delta { seq: *fseq, side, px, qty });
*since += 1;
if *since >= every { *fseq += 1; publish(feeds, snap(book, *fseq)); *since = 0; }
}
// The venue thread body: gateway checks -> sequencer -> engine, fanning out
// private exec reports and public feed messages. Returns the event log.
fn run(cfg: VenueConfig, rx: Receiver<GatewayIn>, exec_tx: Sender<ExecReport>, feeds: Vec<Sender<FeedMsg>>) -> Vec<String> {
let (mut book, mut seq, mut fseq) = (Book::default(), 0u64, 0u64);
let (mut last_px, mut tokens) = (cfg.ref_px, cfg.gw_tokens);
let (mut seen, mut log, mut since) = (HashSet::new(), Vec::new(), 0u64);
for (s, p, q) in &cfg.seed_liq {
book.submit(&NewOrder { client_id: "SEED".into(), side: *s, px: *p, qty: *q });
}
fseq += 1; publish(&feeds, snap(&book, fseq)); // late-joiner contract: snapshot first
for msg in rx.iter() { // the venue's event loop: like `for await` on a stream — sleeps until mail arrives, ends when every sender is gone
thread::sleep(cfg.latency); // artificial venue latency (A fast, B slow)
match msg {
GatewayIn::Cancel { client_id } => {
seq += 1;
let (qty, d) = match book.cancel(&client_id) { Some((s, p, c, l)) => (c, Some((s, p, l))), None => (0, None) };
log.push(format!("{seq} CXL {client_id} qty={qty}"));
// send, not try_send: the private line is reliable — an exec report
// must arrive, so we'd wait rather than drop. Contrast publish() above.
// (.ok() shrugs only if the receiver already shut down.)
exec_tx.send(ExecReport { venue: cfg.id, client_id, exec: ExecType::CancelAck { qty }, seq }).ok();
if let Some((s, p, l)) = d { delta(&book, &feeds, &mut fseq, &mut since, cfg.snap_every, s, p, l); }
}
GatewayIn::CancelAll => { // venue-side kill switch: mass cancel
for (cid, qty) in book.cancel_all() {
seq += 1; log.push(format!("{seq} KILLCXL {cid} qty={qty}"));
exec_tx.send(ExecReport { venue: cfg.id, client_id: cid, exec: ExecType::CancelAck { qty }, seq }).ok();
}
fseq += 1; publish(&feeds, snap(&book, fseq)); since = 0;
}
GatewayIn::New(o) => {
// ---- gateway: per-session risk checks, BEFORE the sequencer ----
let rej = if seen.contains(&o.client_id) { Some("duplicate client_order_id") }
else if o.qty == 0 || o.qty > cfg.max_qty { Some("max order size") }
else if (o.px - last_px).abs() * 10 > last_px { Some("price collar +-10%") }
else if tokens == 0 { Some("gateway rate limit") } else { None };
if let Some(r) = rej {
exec_tx.send(ExecReport { venue: cfg.id, client_id: o.client_id, exec: ExecType::Reject(r.into()), seq: 0 }).ok();
continue; // rejected orders never reach the sequencer: no seq consumed
}
seen.insert(o.client_id.clone());
tokens -= 1;
// ---- sequencer: THE total-order point for this venue ----
seq += 1;
log.push(format!("{seq} NEW {} {:?} {}@{}", o.client_id, o.side, o.qty, o.px));
exec_tx.send(ExecReport { venue: cfg.id, client_id: o.client_id.clone(), exec: ExecType::Ack, seq }).ok();
// ---- matching engine ----
let out = book.submit(&o);
for (px, qty) in &out.taker_fills {
last_px = *px; seq += 1;
log.push(format!("{seq} FILL {} {qty}@{px}", o.client_id));
exec_tx.send(ExecReport { venue: cfg.id, client_id: o.client_id.clone(), exec: ExecType::Fill { px: *px, qty: *qty }, seq }).ok();
}
for (cid, px, qty) in &out.maker_fills {
if cid == "SEED" { continue; }
seq += 1;
exec_tx.send(ExecReport { venue: cfg.id, client_id: cid.clone(), exec: ExecType::Fill { px: *px, qty: *qty }, seq }).ok();
}
for (s, p, l) in out.deltas { delta(&book, &feeds, &mut fseq, &mut since, cfg.snap_every, s, p, l); }
}
}
}
log // event log, returned at shutdown for the determinism check
}
}Step 4 — Broker adapters: normalized book + gap recovery
One adapter thread per venue, each producing a normalized book: both venues’ feeds reduced to the same plain price→qty maps, so the SOR can compare venues without caring whose wire format the update arrived in. lag simulates a slow consumer; the bounded channel it reads is the “socket buffer”. The gap-then-snapshot path in the diagram is exactly ch25’s recovery protocol.
One new idiom here, and it deserves a picture. Each normalized book is a whiteboard: the adapter thread writes price updates on it, the SOR thread reads it to plan routes. Arc means both hold a handle to the same whiteboard — not photocopies; when the adapter writes, the SOR’s next glance sees it. Mutex is the single marker tied to the board: you must hold the marker to touch the board at all — even to read it reliably, since glancing mid-erase shows half-updated numbers — and lock() grabs the marker, waiting if the other thread has it. TypeScript never needs this because one event loop owns all the data; with real threads that guarantee has to be built, and Arc<Mutex<…>> is the smallest way to build it.
══ feed channel, bounded (thread boundary — prod: multicast socket buffer) ══
ADAPTER THREAD │ SOR THREAD
(1) feed_rx: msg arrives (after `lag`) │
(2) book.lock() — take the marker ──────────┼── shared state (lock): NormBook,
(3) Delta{seq}: contiguous? seq == last+1? │ Arc<Mutex<…>> — one board, both
│ no ─▶ gaps += 1, recovering = true │ threads hold handles to it
│ book UNTRUSTED — deltas now │
▼ yes SKIPPED until a snapshot │
(4) apply delta: bids/asks insert/remove │ (6) plan() takes the same lock:
(5) Snapshot: replace book wholesale, │ mid() / ladder() / healthy()
recovering = false ◀── the recovery │ — reads the very board the
path re-enters trusted state HERE │ adapter writes, never a copy
#![allow(unused)]
fn main() {
// src/adapter.rs
use crate::types::*;
use crossbeam_channel::Receiver;
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
/// Broker-side normalized view of one venue, built from the public feed.
pub struct NormBook {
pub bids: BTreeMap<Px, Qty>, pub asks: BTreeMap<Px, Qty>,
pub last_seq: Seq, pub gaps: u32, pub recovering: bool, pub last_update: Instant,
}
impl NormBook {
// A fresh book starts `recovering`: untrusted until the first snapshot lands.
pub fn new() -> Self {
NormBook { bids: BTreeMap::new(), asks: BTreeMap::new(), last_seq: 0, gaps: 0, recovering: true, last_update: Instant::now() }
}
// Midpoint of best bid and best ask. The `?` works like optional chaining `?.`:
// if either side of the book is empty there's no best price, so the answer is None.
pub fn mid(&self) -> Option<f64> {
Some((*self.bids.keys().next_back()? + *self.asks.keys().next()?) as f64 / 2.0)
}
// One side's resting levels as (px, qty) — the ladder the SOR's cost walk consumes.
pub fn ladder(&self, resting: Side) -> Vec<(Px, Qty)> { // best price first
match resting {
Side::Sell => self.asks.iter().map(|(p, q)| (*p, *q)).collect(),
Side::Buy => self.bids.iter().rev().map(|(p, q)| (*p, *q)).collect(),
}
}
// Trustworthy feed = no unresolved gap and updated recently.
pub fn healthy(&self, max_age: Duration) -> bool { !self.recovering && self.last_update.elapsed() <= max_age }
}
/// Adapter thread: drain one venue's feed onto `book`, the whiteboard shared with
/// the SOR (Arc = both hold the SAME board, Mutex = the one marker you must hold).
/// In production this thread reads a multicast socket. `lag` = slow-consumer simulation.
pub fn spawn(venue: VenueId, rx: Receiver<FeedMsg>, book: Arc<Mutex<NormBook>>, lag: Option<Duration>) -> JoinHandle<()> {
thread::spawn(move || {
for msg in rx.iter() {
if let Some(d) = lag { thread::sleep(d); }
// Grab the whiteboard marker; waits if the SOR is mid-read. (unwrap: if a
// holder crashed mid-write the lock is "poisoned" — crash too, don't trust
// a half-updated board.)
let mut b = book.lock().unwrap();
match msg {
FeedMsg::Snapshot { seq, bids, asks } => {
b.bids = bids.into_iter().map(|l| (l.px, l.qty)).collect();
b.asks = asks.into_iter().map(|l| (l.px, l.qty)).collect();
if b.recovering && b.last_seq > 0 { println!(" [adapter {venue:?}] recovered via snapshot seq={seq}"); }
b.last_seq = seq; b.recovering = false; b.last_update = Instant::now();
}
FeedMsg::Delta { seq, side, px, qty } => {
// We number-check every incoming letter. Letter 7 after letter 5
// means 6 is lost in the mail — every conclusion drawn from this
// board is suspect until a fresh snapshot replaces it wholesale.
if seq != b.last_seq + 1 && !b.recovering {
b.gaps += 1; b.recovering = true; // stop applying deltas: the book is untrusted
println!(" [adapter {venue:?}] GAP: expected seq {} got {seq} — waiting for snapshot", b.last_seq + 1);
}
b.last_seq = seq;
if b.recovering { continue; }
let m = match side { Side::Buy => &mut b.bids, Side::Sell => &mut b.asks };
if qty == 0 { m.remove(&px); } else { m.insert(px, qty); }
b.last_update = Instant::now();
}
}
}
})
}
}Step 5 — SOR + parent-order manager
This is the biggest block in the lab, and it does exactly five things. Read them here first — the code is just these five, in order:
The frame for 1–4 is shopping across two stores: one has lower sticker prices but a service fee and a slow checkout, the other is pricier but quick. You compare what it costs to walk out with the goods, not stickers — and you’ll happily buy part of the list at each store.
- Cost model — each visible level is priced as
price + per-venue fee + latency penalty: sticker plus that store’s fee plus its checkout queue. “Cheapest” means all-in cost. - Latency penalty from a measured ack-RTT EWMA — an EWMA (exponentially weighted moving average) is a running estimate that each new reading nudges 30% toward what just happened (
0.7 × old + 0.3 × new); older readings fade geometrically — never gone, just fainter. Every ack round-trip is one reading, so the penalty tracks what the venue is doing now, not its spec sheet. - Cross-venue merge — merge both venues’ eligible levels into ONE list, sorted by that all-in cost.
- Marginal-depth walk — walk the merged list cheapest-first, taking quantity until the parent order is covered. Splitting isn’t a special case: the walk naturally takes from both venues the moment one venue’s next level is dearer than the other’s best remaining.
- Health demotion and the penalty box — venues with a gapped or stale feed are excluded before pricing; venues that rejected this parent are penalty-boxed for the next pass.
Worked through with Step 6’s config, buying 500: B’s 150@10001 is cheapest, B’s 200@10003 next; then A’s 400@10004 beats B’s 300@10005, so the walk finishes there — 500 shares split 350 to B, 150 to A. That is the exact split you’ll see in scenario (a).
Two vocabulary bridges for the code. This is ch26’s parent/child split: the parent is the client’s whole order; the children are the venue-sized slices the SOR cuts it into. And a child is terminal when it is filled, canceled, or rejected — no state left at the venue that could still execute. cancel_children is the double-fill guard from ch26: re-route only after every child is terminal, because a merely sent cancel can still lose the race to a fill already in flight.
Read the code in this order: plan (the five mechanisms above), apply_exec (how acks, fills, rejects, and cancel-acks update state — including the RTT EWMA), execute_parent (the 3-pass send → collect → re-plan loop), then cancel_children (the guard). This is one parent order’s whole life, every thread boundary marked:
time ↓ BROKER / SOR thread │ VENUE A thread │ VENUE B thread
(1) execute_parent: arrival_mid() stamped — TCA benchmark, BEFORE anything sends
(2) plan(): lock both books, merge ladders by all-in cost, walk → child slices
(3) v.tx.send(New P1-C1) ═════════════════▶ gw→seq→engine │
(4) v.tx.send(New P1-C2) ═════════════════╪═══════════════════▶ gw→seq→engine
══ order channels (prod: one TCP order-entry session per venue) ══
(5) collect(): exec_rx.recv_timeout loop │ │
◀═══════════ Ack ═════════════════════╡ │
◀═══════════ Fill{px,qty} ════════════╡ │
◀═══════════ Ack, Fill ═══════════════╪═══════════════════╡
apply_exec: Ack→EWMA, Fill→leaves−=qty │
══ shared exec channel (prod: exec reports on each session) ══
(6) remaining > 0 → cancel before re-route: │
(7) cancel_children: tx.send(Cancel) ═════▶ book.cancel │
◀═══════════ CancelAck{unfilled} ═════╡ │
only NOW may the remainder move — the double-fill guard │
(8) re-plan: back to (2), ≤ 3 passes │ │
(9) remaining == 0 → report_tca(): avg fill px vs arrival mid │
#![allow(unused)]
fn main() {
// src/sor.rs
use crate::{adapter::NormBook, types::*};
use crossbeam_channel::{Receiver, Sender};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
const LAT_PENALTY_PER_MS: f64 = 0.02; // ticks of adverse-selection cost per ms of ack RTT
pub struct VenueLink {
pub id: VenueId, pub tx: Sender<GatewayIn>,
pub book: Arc<Mutex<NormBook>>, pub fee_ticks: f64, pub rtt_ms: f64, // EWMA of measured ack RTT
}
pub struct Sor {
pub venues: Vec<VenueLink>,
pub exec_rx: Receiver<ExecReport>,
next_child: u64,
open_children: HashMap<String, (VenueId, Qty)>, // child -> (venue, leaves)
inflight: HashMap<String, Instant>, // child -> send time (for RTT)
rejected: HashSet<VenueId>, // penalty box, per parent
cur_fills: Vec<(VenueId, Px, Qty)>, // fills for the current parent
}
impl Sor {
// Fresh SOR over these venue links, reading all exec reports from one shared inbox.
pub fn new(venues: Vec<VenueLink>, exec_rx: Receiver<ExecReport>) -> Self {
Sor { venues, exec_rx, next_child: 0, open_children: HashMap::new(),
inflight: HashMap::new(), rejected: HashSet::new(), cur_fills: Vec::new() }
}
// Average mid across healthy venues — the TCA benchmark taken at order arrival.
fn arrival_mid(&self) -> f64 {
let mids: Vec<f64> = self.venues.iter().filter_map(|v| {
let b = v.book.lock().unwrap();
if b.healthy(Duration::from_secs(2)) { b.mid() } else { None }
}).collect();
if mids.is_empty() { 0.0 } else { mids.iter().sum::<f64>() / mids.len() as f64 }
}
// Step 5's mechanisms 1-4 as code: merge every eligible level from every
// healthy venue into one list, priced all-in (px + fee + latency penalty), and
// walk it cheapest-first. Returns child slices as (venue index, limit px, qty).
fn plan(&self, side: Side, qty: Qty, limit: Px) -> Vec<(usize, Px, Qty)> {
let sgn = if side == Side::Buy { 1.0 } else { -1.0 }; // flips sell prices so "smaller = better" holds for both sides
let mut levels: Vec<(f64, usize, Px, Qty)> = Vec::new();
for (i, v) in self.venues.iter().enumerate() {
if self.rejected.contains(&v.id) { continue; }
let b = v.book.lock().unwrap();
if !b.healthy(Duration::from_secs(2)) {
println!(" [sor] venue {:?} unhealthy (gap/stale feed) — demoted", v.id);
continue;
}
let ladder = match side { Side::Buy => b.ladder(Side::Sell), Side::Sell => b.ladder(Side::Buy) };
for (px, q) in ladder {
let within = if side == Side::Buy { px <= limit } else { px >= limit };
if within { levels.push((sgn * px as f64 + v.fee_ticks + v.rtt_ms * LAT_PENALTY_PER_MS, i, px, q)); }
}
}
levels.sort_by(|x, y| x.0.partial_cmp(&y.0).unwrap()); // floats aren't fully ordered in Rust (NaN), hence partial_cmp
let (mut need, mut alloc) = (qty, HashMap::new()); // venue idx -> (worst px, qty)
for (_, i, px, q) in levels {
if need == 0 { break; }
let take = need.min(q);
let e = alloc.entry(i).or_insert((px, 0));
e.0 = px; e.1 += take; need -= take; // walk is per-venue price-ordered, so last px = child limit
}
let mut out: Vec<(usize, Px, Qty)> = alloc.into_iter().map(|(i, (px, q))| (i, px, q)).collect();
out.sort_by_key(|c| c.0);
out
}
// Fold one exec report into SOR state: Ack feeds the RTT estimate, Fill burns
// down leaves, Reject penalty-boxes the venue, CancelAck retires the child.
fn apply_exec(&mut self, r: &ExecReport) {
match &r.exec {
ExecType::Ack => if let Some(t0) = self.inflight.remove(&r.client_id) {
let ms = t0.elapsed().as_secs_f64() * 1e3;
// The EWMA update — nudge the running estimate 30% toward what just
// happened: keep 70% of the old value, blend in 30% of this reading.
// Old readings fade geometrically; the penalty tracks the venue *now*.
if let Some(v) = self.venues.iter_mut().find(|v| v.id == r.venue) { v.rtt_ms = 0.7 * v.rtt_ms + 0.3 * ms; }
},
ExecType::Fill { px, qty } => if let Some(e) = self.open_children.get_mut(&r.client_id) {
e.1 = e.1.saturating_sub(*qty); // subtract, flooring at zero — an unsigned count can't go negative, and wrapping would turn an over-fill into a huge bogus "leaves"
let done = e.1 == 0;
if done { self.open_children.remove(&r.client_id); }
self.cur_fills.push((r.venue, *px, *qty));
println!(" [fill] {:?} {} {qty}@{px}", r.venue, r.client_id);
},
ExecType::Reject(reason) => {
println!(" [sor] venue {:?} REJECT {}: {reason}", r.venue, r.client_id);
self.open_children.remove(&r.client_id);
self.inflight.remove(&r.client_id);
self.rejected.insert(r.venue);
}
ExecType::CancelAck { qty } => {
if *qty > 0 { println!(" [sor] cancel-ack {}: {qty} confirmed unfilled", r.client_id); }
self.open_children.remove(&r.client_id);
}
}
}
// Wait for exec reports until every child in this batch is terminal — or things go quiet.
fn collect(&mut self, batch: &[String]) {
while batch.iter().any(|c| self.open_children.contains_key(c)) {
// recv_timeout: await the next report, but give up after 250 ms of silence — never hang on a quiet channel
match self.exec_rx.recv_timeout(Duration::from_millis(250)) {
Ok(r) => self.apply_exec(&r),
Err(_) => return, // quiet: remaining children are resting at a venue
}
}
}
/// Double-fill guard: cancel, then WAIT for CancelAck (or a racing Fill)
/// for every child before the caller may re-route the remainder anywhere else.
pub fn cancel_children(&mut self, batch: &[String]) {
let pending: Vec<String> = batch.iter().filter(|c| self.open_children.contains_key(*c)).cloned().collect();
for cid in &pending {
let venue = self.open_children[cid].0;
let v = self.venues.iter().find(|v| v.id == venue).unwrap();
v.tx.send(GatewayIn::Cancel { client_id: cid.clone() }).unwrap(); // unwrap: a dead venue thread is a bug worth crashing on
}
let deadline = Instant::now() + Duration::from_secs(1);
while pending.iter().any(|c| self.open_children.contains_key(c)) && Instant::now() < deadline {
if let Ok(r) = self.exec_rx.recv_timeout(Duration::from_millis(200)) { self.apply_exec(&r); }
}
}
/// Run one parent order: up to 3 passes of plan -> send children -> collect ->
/// cancel-and-wait -> re-plan, then the TCA report.
pub fn execute_parent(&mut self, pid: &str, side: Side, qty: Qty, limit: Px) {
let mid = self.arrival_mid();
self.cur_fills.clear();
self.rejected.clear();
println!("[parent {pid}] {side:?} {qty} limit {limit} | arrival mid {mid:.1}");
let mut remaining = qty;
for _pass in 0..3 {
if remaining == 0 { break; }
let plan = self.plan(side, remaining, limit);
if plan.is_empty() { println!(" [sor] no eligible liquidity within limit — leaving {remaining} unfilled"); break; }
let mut batch = Vec::new();
for (i, px, q) in plan {
self.next_child += 1;
let cid = format!("{pid}-C{}", self.next_child);
let v = &self.venues[i];
println!(" [sor] child {cid} -> venue {:?}: {q}@{px} (fee {:.1}t, rtt {:.1}ms)", v.id, v.fee_ticks, v.rtt_ms);
self.open_children.insert(cid.clone(), (v.id, q));
self.inflight.insert(cid.clone(), Instant::now());
v.tx.send(GatewayIn::New(NewOrder { client_id: cid.clone(), side, px, qty: q })).unwrap();
batch.push(cid);
}
self.collect(&batch);
let filled: Qty = self.cur_fills.iter().map(|f| f.2).sum();
remaining = qty - filled.min(qty);
if remaining > 0 {
println!(" [sor] {remaining} unfilled — cancel any resting children, wait for acks, re-plan");
self.cancel_children(&batch);
}
}
self.report_tca(pid, side, mid);
}
// Print the TCA summary: total filled, average price, slippage vs arrival mid, per-venue split.
fn report_tca(&self, pid: &str, side: Side, mid: f64) {
let tot: Qty = self.cur_fills.iter().map(|f| f.2).sum();
if tot == 0 { println!("[parent {pid}] filled 0"); return; }
let notional: f64 = self.cur_fills.iter().map(|f| f.1 as f64 * f.2 as f64).sum();
let sgn = if side == Side::Buy { 1.0 } else { -1.0 };
println!("[parent {pid}] filled {tot} avg {:.2} | slippage vs arrival mid: {:+.2} ticks",
notional / tot as f64, sgn * (notional / tot as f64 - mid));
let mut by: BTreeMap<VenueId, (Qty, f64)> = BTreeMap::new();
for (v, px, q) in &self.cur_fills { let e = by.entry(*v).or_insert((0, 0.0)); e.0 += q; e.1 += *px as f64 * *q as f64; }
for (v, (q, n)) in by { println!(" TCA {v:?}: {q} @ avg {:.2} ({:+.2} ticks vs mid)", n / q as f64, sgn * (n / q as f64 - mid)); }
}
// Send one child straight to a venue — scenario helper for parking resting orders.
pub fn place_resting(&mut self, cid: &str, venue: VenueId, side: Side, px: Px, qty: Qty) {
let v = self.venues.iter().find(|v| v.id == venue).unwrap();
self.open_children.insert(cid.to_string(), (venue, qty));
self.inflight.insert(cid.to_string(), Instant::now());
v.tx.send(GatewayIn::New(NewOrder { client_id: cid.into(), side, px, qty })).unwrap();
}
// Mass-cancel at every venue, then drain acks until open children reconcile to zero.
pub fn kill_switch(&mut self) {
println!("[kill] global cancel — {} open child orders", self.open_children.len());
for v in &self.venues { v.tx.send(GatewayIn::CancelAll).unwrap(); }
let deadline = Instant::now() + Duration::from_secs(1);
while !self.open_children.is_empty() && Instant::now() < deadline {
if let Ok(r) = self.exec_rx.recv_timeout(Duration::from_millis(200)) { self.apply_exec(&r); }
}
println!("[kill] reconcile: open child orders = {} (must be 0)", self.open_children.len());
assert!(self.open_children.is_empty(), "orphan child orders after kill switch");
}
// The parent's fills, packaged as the JSON report a client would receive.
pub fn client_report(&self, parent: &str) -> String {
let fills: Vec<FillLine> = self.cur_fills.iter().map(|(v, px, q)| FillLine { venue: *v, px: *px, qty: *q }).collect();
let filled: Qty = fills.iter().map(|f| f.qty).sum();
let avg = if filled == 0 { 0.0 } else { fills.iter().map(|f| (f.px * f.qty as i64) as f64).sum::<f64>() / filled as f64 };
// serde_json::to_string: Serialize-derived struct -> JSON text, one call
serde_json::to_string(&ClientReport { parent: parent.into(), filled, avg_px: avg, fills }).unwrap()
}
}
}Step 6 — Wiring + four scenarios
main.rs is wiring plus four scripted scenarios. The map before you scroll:
- (a) baseline — proves the cross-venue split from Step 5’s worked example, plus the cancel-ack guard and an idempotency probe (replaying an already-used child ID).
- (b) feed lag — starves B’s feed queue until the venue drops deltas; the adapter gap-detects, the SOR routes around B, and a periodic snapshot repairs the book.
- (c) rate limit — exhausts A’s gateway token budget; the fourth parent is rejected pre-sequencer and spills to B on the re-plan.
- (d) kill switch — pulls the global cancel mid-parent and proves reconciliation reaches zero open orders.
The plumbing: build wires, per venue, one bounded feed channel (a bounded queue = a finite socket buffer), one adapter thread, and one VenueLink for the SOR. b_cap and b_lag are scenario (b)’s knobs — the size of B’s feed queue and how slowly B’s adapter drains it. Each venue’s rtt_ms starts seeded at 2.0 * lat — deliberately pessimistic (twice the configured one-way latency) so the EWMA has something sane to correct once real acks are measured. shutdown works by dropping the SOR: that drops the order-entry senders, so the venues drain and exit, their feed channels close, and the adapters exit — a clean cascade with no shutdown flag.
// src/main.rs
mod adapter; mod book; mod sor; mod types; mod venue;
use crossbeam_channel::{bounded, unbounded};
use sor::{Sor, VenueLink};
use std::sync::{Arc, Mutex};
use std::thread::{sleep, JoinHandle};
use std::time::Duration;
use types::*;
use venue::VenueConfig;
// Shared venue config; the knobs that differ per venue come in as arguments.
fn cfg(id: VenueId, lat_ms: u64, fee_ticks: f64, tokens: u32, seed_liq: Vec<(Side, Px, Qty)>) -> VenueConfig {
VenueConfig { id, ref_px: 10_000, fee_ticks, latency: Duration::from_millis(lat_ms),
max_qty: 1_000, gw_tokens: tokens, snap_every: 10, seed_liq }
}
fn cfg_a() -> VenueConfig { // fast + deep, but wide spread and higher fee
cfg(VenueId::A, 2, 0.6, 100, vec![(Side::Buy, 9_996, 400), (Side::Buy, 9_994, 500),
(Side::Sell, 10_004, 400), (Side::Sell, 10_006, 500)])
}
fn cfg_b() -> VenueConfig { // slow + thin, but tight spread and lower fee
cfg(VenueId::B, 20, 0.2, 100, vec![(Side::Buy, 9_999, 150), (Side::Buy, 9_997, 200),
(Side::Sell, 10_001, 150), (Side::Sell, 10_003, 200), (Side::Sell, 10_005, 300)])
}
struct Market { sor: Sor, venues: Vec<JoinHandle<Vec<String>>>, adapters: Vec<JoinHandle<()>> }
// Wire the whole market: per venue, one feed channel + adapter thread + VenueLink;
// one shared exec-report channel feeding the SOR.
fn build(cfgs: [VenueConfig; 2], b_cap: usize, b_lag: Option<Duration>) -> Market {
let (exec_tx, exec_rx) = unbounded();
let (mut links, mut vj, mut aj) = (Vec::new(), Vec::new(), Vec::new());
let [ca, cb] = cfgs;
for (c, cap, lag) in [(ca, 64usize, None), (cb, b_cap, b_lag)] {
let (ftx, frx) = bounded(cap); // the mail slot from Step 3, `cap` letters deep — this IS the "socket buffer" the venue drops on
let (id, fee, lat) = (c.id, c.fee_ticks, c.latency.as_millis() as f64);
let h = venue::spawn(c, exec_tx.clone(), vec![ftx]);
let book = Arc::new(Mutex::new(adapter::NormBook::new()));
aj.push(adapter::spawn(id, frx, book.clone(), lag));
links.push(VenueLink { id, tx: h.tx, book, fee_ticks: fee, rtt_ms: 2.0 * lat });
vj.push(h.join);
}
sleep(Duration::from_millis(60)); // let adapters ingest the initial snapshots
Market { sor: Sor::new(links, exec_rx), venues: vj, adapters: aj }
}
// Drop-driven teardown; the joins prove every thread actually exited.
fn shutdown(m: Market) {
drop(m.sor); // drops order-entry senders -> venues drain + exit -> feeds close -> adapters exit
for j in m.venues { j.join().unwrap(); } // join: block until the thread exits and hand back its return value; unwrap re-raises any panic it died with
for j in m.adapters { j.join().unwrap(); }
}
// (a) Baseline: the Step 5 split, an idempotency probe, and the cancel-ack re-route guard.
fn scenario_a() {
println!("\n=== (a) baseline: split across A+B, TCA vs arrival mid ===");
let mut m = build([cfg_a(), cfg_b()], 64, None);
m.sor.execute_parent("P1", Side::Buy, 500, 10_006);
println!("{}", m.sor.client_report("P1"));
// idempotency probe: replay child P1-C1's id at venue A -> gateway rejects the duplicate
m.sor.venues[0].tx.send(GatewayIn::New(NewOrder { client_id: "P1-C1".into(), side: Side::Buy, px: 10_004, qty: 10 })).unwrap();
if let Ok(r) = m.sor.exec_rx.recv_timeout(Duration::from_millis(300)) { println!("replay P1-C1 -> {:?}", r.exec); }
// partial fill + re-route: a resting child must be CANCELED AND ACKED before re-routing
m.sor.place_resting("P2-R1", VenueId::A, Side::Buy, 9_998, 200); // below the ask: rests
sleep(Duration::from_millis(50));
println!("[parent P2] re-route: cancel resting child, wait for cancel-ack, only then send remainder");
m.sor.cancel_children(&["P2-R1".into()]);
m.sor.execute_parent("P2", Side::Buy, 200, 10_006); // safe: venue confirmed 200 unfilled
shutdown(m);
}
// (b) Slow subscriber: overflow B's feed queue, watch gap -> demote -> snapshot recovery.
fn scenario_b() {
println!("\n=== (b) venue B feed lags: gap -> demote -> route around -> snapshot recovery ===");
let mut m = build([cfg_a(), cfg_b()], 2, Some(Duration::from_millis(60))); // tiny buffer + slow adapter
for i in 0..30 { // churn on B overflows the lagging subscriber's queue -> venue drops deltas
m.sor.venues[1].tx.send(GatewayIn::New(NewOrder { client_id: format!("N{i}"), side: Side::Sell, px: 10_006 + (i % 3), qty: 10 })).unwrap();
}
sleep(Duration::from_millis(500)); // by now B's adapter has hit the gap and is in recovery
m.sor.execute_parent("P1", Side::Buy, 300, 10_006);
for i in 0..12 { // slow churn: once the adapter drains, a periodic snapshot repairs the book
m.sor.venues[1].tx.send(GatewayIn::New(NewOrder { client_id: format!("R{i}"), side: Side::Sell, px: 10_009, qty: 5 })).unwrap();
sleep(Duration::from_millis(80));
}
sleep(Duration::from_millis(800));
shutdown(m);
}
// (c) Token-bucket burst: A's gateway runs dry; the re-plan spills to B.
fn scenario_c() {
println!("\n=== (c) burst: venue A gateway rate-limits, SOR spills to B ===");
let a = cfg(VenueId::A, 2, 0.1, 3, vec![(Side::Buy, 9_998, 600), (Side::Sell, 10_001, 600)]);
let b = cfg(VenueId::B, 20, 0.2, 100, vec![(Side::Buy, 9_997, 600), (Side::Sell, 10_004, 600)]);
let mut m = build([a, b], 64, None);
for p in 1..=4 { m.sor.execute_parent(&format!("P{p}"), Side::Buy, 100, 10_008); }
shutdown(m);
}
// (d) Kill switch with live children at both venues; reconciliation must hit zero.
fn scenario_d() {
println!("\n=== (d) kill switch mid-parent: global cancel, zero orphans ===");
let mut m = build([cfg_a(), cfg_b()], 64, None);
m.sor.place_resting("K-1", VenueId::A, Side::Buy, 9_997, 200);
m.sor.place_resting("K-2", VenueId::B, Side::Buy, 9_998, 150);
sleep(Duration::from_millis(80)); // children are live at both venues
m.sor.kill_switch();
shutdown(m);
}
// A seeded pseudo-random generator: same seed, same sequence, forever, on any
// machine. That predictability is the whole point — the determinism check must be
// able to feed the venue an IDENTICAL order script twice and diff the logs.
struct Rng(u64); // xorshift64: 8 bytes of state, no rand crate
impl Rng {
// seed.max(1): xorshift state must never be 0 — 0 maps to 0 forever.
fn new(seed: u64) -> Self { Rng(seed.max(1)) }
// Each call: three shift-and-XOR rounds smear the state's bits around. It looks
// random but is pure arithmetic — nothing reads a clock or an OS entropy pool.
fn next(&mut self) -> u64 { let mut x = self.0; x ^= x << 13; x ^= x >> 7; x ^= x << 17; self.0 = x; x }
}
// Drive one venue with 40 seeded pseudo-random orders; return its event log.
fn scripted_venue(seed: u64) -> Vec<String> {
let (etx, _erx) = unbounded();
let (ftx, _frx) = bounded(4_096);
let h = venue::spawn(cfg_a(), etx, vec![ftx]);
let mut rng = Rng::new(seed);
for i in 0..40 {
let side = if rng.next() % 2 == 0 { Side::Buy } else { Side::Sell };
let (px, qty) = (9_995 + (rng.next() % 11) as Px, 10 + rng.next() % 90);
h.tx.send(GatewayIn::New(NewOrder { client_id: format!("D{i}"), side, px, qty })).unwrap();
}
drop(h.tx);
h.join.join().unwrap()
}
// Same seed, two runs, byte-identical logs — the event-sourcing claim, checked.
fn determinism() {
let (r1, r2) = (scripted_venue(42), scripted_venue(42));
assert_eq!(r1, r2);
println!("\n[determinism] venue A replayed on seed 42: {} log lines, byte-identical", r1.len());
}
// Everything in sequence; each scenario builds and tears down its own market.
fn main() { scenario_a(); scenario_b(); scenario_c(); scenario_d(); determinism(); println!("\nall scenarios complete"); }cargo run --release (~5 s total). Timings and RTT figures below will wobble on your machine; prices, quantities, and event ordering will not — the scripted sleeps hold these scenarios’ broker outcomes stable, though in general (Step 7) the broker side wobbles too.
Scenario (a) — the merged cost walk plays out exactly as Step 5’s worked example: B’s best prices win first even though B is slow, and once B’s cheap levels run out, A’s bigger size takes over. (B’s RTT shows as 40 ms because we seeded the estimate at twice its 20 ms configured latency — no ack has been measured yet.) The summary at the end is the TCA report from ch26: average fill price against the mid-price at the moment the order arrived (the “arrival mid”), the difference being the slippage:
=== (a) baseline: split across A+B, TCA vs arrival mid ===
[parent P1] Buy 500 limit 10006 | arrival mid 10000.0
[sor] child P1-C1 -> venue A: 150@10004 (fee 0.6t, rtt 4.0ms)
[sor] child P1-C2 -> venue B: 350@10003 (fee 0.2t, rtt 40.0ms)
[fill] A P1-C1 150@10004
[fill] B P1-C2 150@10001
[fill] B P1-C2 200@10003
[parent P1] filled 500 avg 10002.70 | slippage vs arrival mid: +2.70 ticks
TCA A: 150 @ avg 10004.00 (+4.00 ticks vs mid)
TCA B: 350 @ avg 10002.14 (+2.14 ticks vs mid)
{"parent":"P1","filled":500,"avg_px":10002.7,"fills":[{"venue":"A","px":10004,"qty":150},{"venue":"B","px":10001,"qty":150},{"venue":"B","px":10003,"qty":200}]}
replay P1-C1 -> Reject("duplicate client_order_id")
[parent P2] re-route: cancel resting child, wait for cancel-ack, only then send remainder
[sor] cancel-ack P2-R1: 200 confirmed unfilled
[parent P2] Buy 200 limit 10006 | arrival mid 10001.0
[sor] child P2-C3 -> venue A: 200@10004 (fee 0.6t, rtt 18.7ms)
[fill] A P2-C3 200@10004
[parent P2] filled 200 avg 10004.00 | slippage vs arrival mid: +3.00 ticks
TCA A: 200 @ avg 10004.00 (+3.00 ticks vs mid)
(Notice P2’s RTT estimate jumped to ~19 ms: the resting child’s ack sat unread in the broker’s inbox during our 50 ms sleep, so when we finally read it, the measured “round trip” included our own nap — and the EWMA honestly absorbed that. Measured latency includes your own processing delays — a real SOR lesson for free.)
Scenario (b) — the overflow is pure arithmetic: B’s adapter takes 60 ms per message (b_lag), the burst of 30 orders produces deltas far faster than that, and the queue between them holds only 2 (b_cap). The queue fills in the first burst, so the venue’s try_send starts failing — those are the dropped deltas. The venue never blocks; dropping is the multicast contract:
time ↓ BROKER (adapter B + SOR) │ VENUE A │ VENUE B
(1) 30 sell orders ═══════════════════════╪═════════▶ deltas flood the feed
(2) B's feed queue (cap 2) fills → venue try_send DROPS deltas — no blocking
(3) adapter B: seq 8 after 6 → GAP — │ │
recovering = true, book untrusted │ │
(4) P1: plan() → B !healthy() → demoted │ │
(5) child ════════════════════════════════▶ fills 300@10004 — A only
(6) slow churn ═══════════════════════════╪═════════▶ adapter drains at its pace
(7) periodic Snapshot ◀═══════════════════╪═════════╡ book rebuilt wholesale
(8) recovering = false → B healthy again, back in the next merge
=== (b) venue B feed lags: gap -> demote -> route around -> snapshot recovery ===
[adapter B] GAP: expected seq 6 got 8 — waiting for snapshot
[parent P1] Buy 300 limit 10006 | arrival mid 10000.0
[sor] venue B unhealthy (gap/stale feed) — demoted
[sor] child P1-C1 -> venue A: 300@10004 (fee 0.6t, rtt 4.0ms)
[fill] A P1-C1 300@10004
[parent P1] filled 300 avg 10004.00 | slippage vs arrival mid: +4.00 ticks
TCA A: 300 @ avg 10004.00 (+4.00 ticks vs mid)
[adapter B] recovered via snapshot seq=45
Scenario (c) — A is cheaper here, so the first three parents drain its gateway token budget — a token bucket: each venue’s gateway allows a fixed number of new orders per window, and this scenario’s A was configured with only 3 tokens (production gateways refill the bucket on a timer; the lab deliberately never refills within a run). The fourth parent is rejected pre-sequencer and spills to B on the re-plan — the next pass of execute_parent’s loop from Step 5: cancel anything resting, wait for the acks, plan again with A in the penalty box:
time ↓ BROKER (SOR) │ VENUE A (3 tokens) │ VENUE B
(1) P1..P3 children ══════════════════════▶ tokens 3→2→1→0, │
(2) P4 child ═════════════════════════════▶ bucket empty │
(3) ◀══ Reject("gateway rate limit") ═════╡ pre-sequencer: │
apply_exec: A → penalty box (rejected)│ no seq consumed │
(4) cancel_children (nothing resting), then re-plan │
(5) plan(): A skipped → child ════════════╪════════════════════▶ gw→seq→engine
(6) ◀═══════════ Ack + Fill 100@10004 ════╪════════════════════╡
=== (c) burst: venue A gateway rate-limits, SOR spills to B ===
[parent P1] Buy 100 limit 10008 | arrival mid 10000.0
[sor] child P1-C1 -> venue A: 100@10001 (fee 0.1t, rtt 4.0ms)
[fill] A P1-C1 100@10001
[parent P1] filled 100 avg 10001.00 | slippage vs arrival mid: +1.00 ticks
...P2, P3 identical, filled at A...
[parent P4] Buy 100 limit 10008 | arrival mid 10000.0
[sor] child P4-C4 -> venue A: 100@10001 (fee 0.1t, rtt 3.1ms)
[sor] venue A REJECT P4-C4: gateway rate limit
[sor] 100 unfilled — cancel any resting children, wait for acks, re-plan
[sor] child P4-C5 -> venue B: 100@10004 (fee 0.2t, rtt 40.0ms)
[fill] B P4-C5 100@10004
[parent P4] filled 100 avg 10004.00 | slippage vs arrival mid: +4.00 ticks
Scenario (d) — kill switch: mass-cancel at both venues, then reconcile the broker’s open-order table against venue acks. Zero means zero:
time ↓ BROKER (SOR) │ VENUE A │ VENUE B
(1) K-1 rests at A, K-2 at B — open_children = 2
(2) kill_switch: tx.send(CancelAll) ══════▶ cancel_all
tx.send(CancelAll) ══════╪═════════▶ cancel_all
(3) ◀══ CancelAck K-1 (200 unfilled) ═════╡ │
(4) ◀══ CancelAck K-2 (150 unfilled) ═════╪═════════╡ each ack retires a child
(5) reconcile: open_children == 0 — venue-CONFIRMED zero, then the assert
=== (d) kill switch mid-parent: global cancel, zero orphans ===
[kill] global cancel — 2 open child orders
[sor] cancel-ack K-1: 200 confirmed unfilled
[sor] cancel-ack K-2: 150 confirmed unfilled
[kill] reconcile: open child orders = 0 (must be 0)
Step 7 — Determinism: one side has it, the other never will
[determinism] venue A replayed on seed 42: 66 log lines, byte-identical
VENUE lane — deterministic BROKER lane — nondeterministic (timing)
run 1: script(seed 42) ─▶ venue ─▶ log₁ threads race: ack interleaving, RTTs,
run 2: script(seed 42) ─▶ venue ─▶ log₂ book-at-plan-time all wobble run to
assert_eq!(log₁, log₂) — byte-compare ✓ run → no replay; keep a timestamped
(same sequenced input ⇒ same event log) DECISION LOG instead
The venue is deterministic because its entire state is a function of the sequenced input: one consumer, no wall-clock in any decision, artificial latency that delays but never reorders — ch13’s event-sourcing claim made concrete, and why real venues can replay a day from the sequenced log for disputes and testing.
The broker side is not deterministic and cannot be made so cheaply: the threads race, so the same parent order can legitimately split differently on two runs.
Real systems don’t fight this; they evidence it: every routing decision is logged with a timestamp and the inputs it saw (the normalized books, health flags, measured RTTs). That decision log — not a replayable world — is what backs a best-execution defense (ch26).
Plain-English recap
- A venue is three stages: gateway (per-session checks: collar, size, duplicate ID, rate limit — all before sequencing), sequencer (one consumer = the total order), engine (pure function of the sequence). Rejects consume no sequence number.
- The venue has two outputs with different contracts: private exec reports (reliable, to you) and a public feed (best-effort, seq-numbered, snapshot every N so anyone can join or recover).
- A feed publisher never blocks on a slow subscriber — it drops. Recovery is the subscriber’s job: detect the seq gap, distrust the book, rebuild from the next snapshot.
- A broker adapter’s product is a normalized book plus a health flag. The SOR consumes both: cost model over healthy venues, demotion for the rest.
- Routing cost is not just price: fees and latency (measured, not assumed) shift the split.
- Re-routing an unfilled child without waiting for the cancel-ack is how you buy the same shares twice. Cancel, wait for the terminal event, then re-route.
- A kill switch is only done when reconciliation says zero open orders — venue-confirmed, not assumed.
Interviewer will ask
“Where does an order become real?” At the sequencer. My lab’s gateway rejects (collar, dup ID, rate limit) happen before sequencing and consume no seq number; from the sequencer onward everything is a deterministic function of the sequence — I verified byte-identical logs on replay.
“Why can’t the feed publisher just block briefly on a slow consumer?” Then one slow subscriber sets the venue’s pace for everyone — an outsider controlling the matching path. I used try_send and dropped; in scenario (b) the venue stayed at full speed while one subscriber went blind and recovered later.
“How does your feed handler recover from a loss?” Contiguous seq check on every delta; on a gap I mark the book untrusted and apply nothing until the next periodic snapshot (published every N deltas), then resume from its seq. Meanwhile the SOR sees recovering and routes around that venue.
“Why cancel-then-wait before re-routing?” Because cancel is a race against a fill in flight. If I re-route while the child might still execute, both can fill — a doubled position. The venue’s CancelAck (or the racing Fill) is the only authority on how much was left; I re-route exactly that confirmed quantity.
“How does your SOR choose venues?” Effective cost per level: price + venue fee + a latency penalty from an EWMA of measured ack RTTs, merged across venues, cheapest marginal depth first — so it splits exactly when one venue’s next level is worse than the other venue’s best remaining. Health-gated: gapped or stale feeds are excluded before pricing.
“What did your kill switch actually prove?” Not that I sent cancels — that reconciliation reached zero: every child the broker believed open was matched by a venue CancelAck. Open state you can’t reconcile is risk you can’t bound.
“Which half is deterministic and why?” Venue: yes — single sequencer, state a pure function of sequenced input; identical logs across runs. Broker: no — independent venue threads race, so decisions vary. The fix isn’t determinism, it’s timestamped decision logging: record what the router saw when it chose, for best-ex evidence.
“Why check duplicate client order IDs at the gateway?” Idempotency for retransmits: if my session resends after a timeout, the dup check turns a would-be double execution into a reject before the sequencer. I demonstrated it by replaying a filled child’s ID and getting Reject("duplicate client_order_id").
Further reading
- Larry Harris, Trading and Exchanges — venue microstructure, order precedence, why the rules look like this.
- Nasdaq TotalView-ITCH 5.0 and OUCH protocol specifications — real seq-numbered feed and order-entry message sets; compare with your
FeedMsg/GatewayIn. - CME MDP 3.0 market data documentation — incremental + snapshot recovery channels in production form.
- FIX Trading Community, FIX protocol specification — ExecutionReport lifecycle (your
ExecTypein the wild). - SEC Regulation NMS (esp. Rule 611) and FINRA Rule 5310 — the regulatory frame behind “best execution evidence”.
The sentences you can now say in interviews truthfully
- “I’ve built a two-venue market end-to-end in Rust — gateway, sequencer, matching engine, seq-numbered feed with snapshot recovery, plus the broker side: adapters, SOR, and a parent-order manager.”
- “I’ve watched a venue drop feed messages on a slow subscriber and implemented the gap-detect-then-resnapshot recovery on the consuming side.”
- “My SOR splits parents using effective cost — price plus fees plus a measured latency penalty — and demotes venues on feed-health signals.”
- “I’ve implemented the cancel-then-wait-for-ack guard and can explain exactly which race it closes.”
- “I’ve verified a sequenced venue is byte-for-byte replayable, and I can explain why the multi-venue broker side isn’t — and what decision logging does about it.”
- “I’ve built a kill switch that proves completion by reconciling open-order state to zero against venue acks.”