Multi-Instrument Engine

The partition key is the instrument. Every operation that touches exactly one book (new order, cancel, replace, the match itself, that book's market data) goes to the book's shard, and it's processed in the order that shard receives it. Most of the message volume falls in that category.

Then list the operations that cross partitions, because each needs a deliberate design:

• Per-account risk. An account trades ES on shard 0 and NQ on shard 2 against one gross-exposure limit. Who knows the total?
• Sessions. One client connection sends orders for many instruments. Its acks and fills come back from several shards and interleave on one TCP stream.
• Mass cancel and kill switches. "Cancel everything for account X" has to reach every shard and be acknowledged by all of them.
• Spread and implied instruments. A calendar spread's legs are two outright books. Implied matching (as CME does for many futures) needs both legs' books in one place, so co-locate related instruments in the same shard. This is the one cross-instrument interaction you keep inside a partition.
• Market-wide state. Trading halts, circuit breakers and session transitions (open, close, auction) must apply to every shard at a well-defined point in each shard's sequence.
• Global ordering. Consolidated audit trails and some market-data consumers want a single stream. Producing one costs a merge, and the merge has to be deterministic.

Instrument reference data is the table every component consults: symbol, a dense numeric id, tick size, lot size, price bands, trading status, and the shard that owns it. Three rules keep it from becoming a source of bugs:

1. Strings stop at the edge. The gateway resolves "ESZ6" to instrument id 17 once per order (or once per session if clients use ids). From then on everything is an integer, and shard_of[17] is one array load.
2. It's versioned and sequenced. Reference data loads at start of day. Any intraday change (a new instrument, a band adjustment, a halt) arrives as an admin message through the shard's input sequence, so replaying the journal reproduces the same state at the same point.
3. It's deterministic. Never let anything that affects output depend on std::collections::HashMap iteration order or its hash values. The default hasher is SipHash with random keys per process, so two runs, or the primary and its standby (week 20), would disagree. For anything that must be reproducible, use a fixed hash such as FNV-1a, or better, an explicit assignment table.

Assignment is where most first designs go wrong. Hashing symbols to shards is uniform over symbols, but trading volume is wildly non-uniform. A handful of instruments (the front-month equity index future, the most active stocks) can carry a large fraction of all messages, roughly following a power law. Hash four busy symbols onto four shards and three of them can land together, as you'll see in the drill. The practical approach is an explicit table built from measured volume: greedy longest-processing-time (place the heaviest instrument on the least-loaded shard, repeat) gets within a factor of 4/3 of optimal, and the very heaviest instruments get a shard to themselves. Rebuild the table at start of day from yesterday's statistics, and never move a book intraday. That would need a quiesce, a state transfer and a journal handover, all while the market is open.

use std::collections::HashMap;

/// FNV-1a, 64-bit (offset basis 0xcbf29ce484222325, prime 0x100000001b3).
/// Deterministic across processes and runs, unlike std's RandomState.
pub fn fnv1a(s: &str) -> u64 {
    todo!()
}

/// fnv1a(symbol) % shards
pub fn hash_shard(symbol: &str, shards: usize) -> usize {
    todo!()
}

/// Greedy longest-processing-time assignment: visit instruments by weight
/// descending (ties: lower index first) and put each on the least-loaded
/// shard so far (ties: lower shard index). Returns the shard per instrument.
pub fn assign_lpt(weights: &[u64], shards: usize) -> Vec<usize> {
    todo!()
}

/// Load of the busiest shard.
pub fn max_load(weights: &[u64], assignment: &[usize], shards: usize) -> u64 {
    todo!()
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Route {
    pub instrument: u32,
    pub shard: usize,
}

/// Symbols are resolved once, at the edge; the hot path carries dense ids.
pub struct Router {
    by_symbol: HashMap<String, u32>,
    shard_of: Vec<usize>, // indexed by instrument id
}

impl Router {
    /// Instrument id = position in `symbols`; shard from `assignment`.
    pub fn new(symbols: &[&str], assignment: &[usize]) -> Self {
        todo!()
    }

    /// Edge lookup: symbol -> (instrument id, shard). None if unknown.
    pub fn route(&self, symbol: &str) -> Option<Route> {
        todo!()
    }

    /// Hot-path lookup: one array index, no hashing.
    pub fn shard_of(&self, instrument: u32) -> usize {
        todo!()
    }
}

The drill's Router uses a HashMap for the edge lookup, and that's fine: lookup results don't depend on the hasher's keys, only iteration order and hash values do. Which of your structures iterate a HashMap in a way that affects output?

Each shard gets its own sequencer: an input sequence number stamped on every message the shard accepts, and an output sequence on every event it emits. Everything you built for one engine now holds per shard:

• Journals. Each shard appends to its own journal (week 9). Journals don't contend with each other, and recovery replays them in parallel, one per shard.
• Determinism. A shard's state is a pure function of its own journal. Cross-shard effects (an admin halt, a mass cancel) enter as messages in each affected shard's journal, so they're part of its sequence.
• Market data. Each shard publishes on its own channel with its own sequence numbers, and consumers detect gaps per channel. This is the CME model: a gap on one channel triggers recovery for that channel only. A single consolidated feed (the ITCH model) needs a publisher that merges shard outputs and assigns one global sequence. It's simpler for consumers, but it's a serialisation point that every message passes through.

Sometimes you do need one ordered stream: a consolidated audit trail, a drop copy, a single-feed product. Then define the order deterministically from the logs, not from arrival times at some merging thread. Ordering by (timestamp, partition, sequence) is reproducible given the journals. It's only as meaningful as the clocks are synchronised, so stamp from one well-synchronised clock (the TSC on one host, PTP across hosts, week 21) and document the tie-break.

use std::cmp::Reverse;
use std::collections::BinaryHeap;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Event {
    pub partition: usize,
    pub seq: u64,
    pub ts: u64,
}

/// One counter per partition. No shared state between partitions, so each
/// shard can stamp its own events without coordination.
pub struct Sequencer {
    next: Vec<u64>,
}

impl Sequencer {
    pub fn new(partitions: usize) -> Self {
        todo!()
    }

    /// Next sequence number for this partition; each partition starts at 1.
    pub fn stamp(&mut self, partition: usize) -> u64 {
        todo!()
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SeqCheck {
    InOrder,
    /// Already seen, or inside a gap you already reported (recovery's job).
    Stale,
    /// Missing range [from, to]; the message itself is accepted.
    Gap { from: u64, to: u64 },
}

/// Consumer-side gap detection, independently per partition.
pub struct GapDetector {
    expected: Vec<u64>,
}

impl GapDetector {
    pub fn new(partitions: usize) -> Self {
        todo!()
    }

    pub fn check(&mut self, partition: usize, seq: u64) -> SeqCheck {
        todo!()
    }
}

/// Merge per-partition logs (each in seq order, timestamps non-decreasing)
/// into one stream ordered by (ts, partition, seq). Deterministic: the same
/// logs always produce the same stream. A k-way merge with a heap is the
/// natural fit.
pub fn merge(logs: &[Vec<Event>]) -> Vec<Event> {
    todo!()
}

With N gateway threads and M shard threads, the simplest wiring is one MPSC queue per shard. Every gateway pushes into it, and producers contend on its tail with a CAS (week 5). The better default is one SPSC ring per (gateway, shard) pair: N × M rings with no contention and no CAS, each with exactly one producer and one consumer. With 2 gateways and 4 shards that's 8 small rings, which costs nothing. Each shard polls its N inbound rings round-robin with a per-ring batch cap so no gateway starves the others. Replies flow back the same way, one ring per (shard, gateway).

There's a subtle bonus. A single session lives on one gateway thread, so its messages to a given shard travel through one FIFO ring. Per-session order is preserved within each shard, which is exactly the guarantee clients need: a cancel can never overtake the order it cancels.

Three more wiring decisions:

• Backpressure. If a shard's inbound ring is full, the gateway must not spin waiting. That would stall every session on that gateway, including ones trading on idle shards (head-of-line blocking). Reject the order with a "busy" reason, or park it in a small per-shard overflow queue. Either way, count it, because a full ring means that shard is overloaded.
• Placement. Each shard thread gets an isolated core on the NIC's NUMA node (week 30), with its journal buffer and book memory allocated from that node.
• False sharing. Per-shard counters and ring indices live in cache-padded structs (week 4). Four shards incrementing adjacent u64 statistics will share a line and bounce it.

Per-account risk is the hardest. There are three common shapes, each with a different failure mode:

1. Central risk before routing. Every order passes one risk thread that knows every account's total exposure. It's exact and simple, but it's a serialisation point: you've rebuilt the single-threaded bottleneck one hop earlier. That's fine if risk checks are much cheaper than matching (often true), and fatal if not.
2. Partitioned credit. Split each account's limit across shards (say, in proportion to where it traded yesterday). Each shard checks locally with no coordination, and a slow background process moves unused credit between shards. It's fast and deterministic per shard, but an order can be rejected for lack of local credit while the account has plenty in total.
3. Local checks with asynchronous aggregation. Shards check against a conservative local view and publish fills to a global aggregator that can tighten limits or trigger a kill. It's fast, but the global limit can be briefly exceeded, by at most the in-flight volume. Size the headroom for that.

Mass cancel fans out: the gateway sends "cancel all for account X" to every shard holding X's orders. Each shard executes it at a definite point in its own sequence and acknowledges, and the gateway reports completion when all acks are in. Because of the per-ring FIFO, every order the session sent before the mass cancel reaches each shard before it, so nothing slips through. An order sent after the cancel is a new order. Decide whether a kill-switched account's new orders are rejected at the gateway (you want that) or left to each shard.

Market-wide events (halts, circuit breakers, the transition into a closing auction) enter every shard as an admin message. Each shard applies it at its own sequence point, and the journals record exactly which orders were before and after. They won't be simultaneous across shards in wall-clock time, and you don't need them to be. You need each shard's journal to say what happened.

Your engine v1 becomes the shard. Mostly unchanged, it owns a subset of books, its own sequencer, journal and market-data output. Around it you add:

• InstrumentRegistry: reference data, loaded once, immutable for the session, with symbol→id at the edge and an id→shard table built from a weight file (LPT) or an explicit map. The repo's registry is keyed by InstrumentId, so symbols are resolved before it. register(spec) puts an instrument on the shard with the fewest instruments (round robin for a fresh registry), and register_on(spec, shard) places it explicitly. A volume-balanced table is a list of register_on calls you compute yourself.
• ShardedEngine::new(registry, batch_size): spawns one pinned thread per shard and wires the rings. In the repo the threads aren't pinned yet, and the queues are std::sync::mpsc channels.
• A router that resolves ids and pushes to the right queue. In the repo it lives inside ShardedEngine, on the caller's thread: send(&cmd).
• Market-data aggregation: either per-shard channels or a deterministic merge into one stream. Pick one and justify it.