Deterministic Event Sourcing

There are two things you could journal: the inputs (new order, cancel, session end, reference-data change) or the outputs (accepted, trade, rested, cancelled). For rebuilding state, the inputs are enough and are the more fundamental record: outputs are a deterministic consequence of inputs, but the reverse is not true — the events do not record that a cancel arrived for an order that had already filled. Many venues keep both, because outputs are what was told to the world and are needed to answer disputes quickly, but the input journal is what recovery and replication are built on.

The rule that makes it safe: no output leaves the exchange for an input that is not durably recorded (and, with a standby, replicated). Otherwise a crash can leave a client holding an execution report for a trade that the restarted exchange, replaying its journal, never makes.

The sequencer is the single point where concurrent input becomes one ordered stream. Gateways receive orders in parallel; the sequencer stamps each with the next sequence number (and, if the business logic needs time, a timestamp), and from then on everything downstream — matching, journal, standby, market data — consumes the same numbered stream.

Sequence numbers do three jobs:

• Ordering. Command n is processed before n + 1, everywhere.
• Gap detection. A consumer expecting n + 1 that receives n + 3 knows it missed two commands and must recover them before going on (week 10 does this for market data).
• Idempotency. A replica that has applied up to n ignores any re-delivered command with a sequence number at or below n, so retransmission is always safe.

The sequencer is also where fairness is decided: whichever command gets the lower number "arrived first", regardless of which gateway it came through. That is why it must be a single, deterministic component, and why its own availability (week 20) is the hardest part of the design.

A journal is an append-only sequence of frames. Each frame needs:

• a length, so a reader knows where the payload ends and the next frame begins;
• a sequence number, so a reader can detect gaps and duplicates;
• a checksum over everything after it, so a reader can tell a complete, intact frame from a corrupt or half-written one;
• the payload: the encoded command.

The checksum of choice is CRC-32C (Castagnoli). It detects every burst error up to 32 bits and random corruption with probability about 1 − 2⁻³², it is what iSCSI (RFC 3720), ext4 and Btrfs metadata and Kafka record batches use, and modern x86 (SSE4.2) and ARMv8 cores compute it in hardware. It is an error-detecting code, not a cryptographic hash: it protects against accidents, not attackers.

The interesting case is recovery. A crash in the middle of an append leaves a torn write: a frame whose header or payload is only partly on disk. On restart you scan from the start, accept frames while they are complete, intact and consecutive, and truncate at the first one that is not. A short frame at the very end is expected after a crash; a corrupt frame or a sequence gap in the middle of the file is not, and it should stop recovery loudly rather than being skipped.

/// Frame layout, all integers little-endian:
///
///   [0..4)        crc   u32  CRC-32C of bytes [4 .. 16 + len)  (len, seq, payload)
///   [4..8)        len   u32  payload length
///   [8..16)       seq   u64  sequence number
///   [16..16+len)  payload
pub const HEADER: usize = 16;
pub const MAX_PAYLOAD: usize = 1 << 20;

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum FrameError {
    /// Fewer bytes than the frame needs. At the end of a journal this is a
    /// torn write; anywhere else it is a bug.
    Incomplete,
    /// `len` > MAX_PAYLOAD — a corrupt header; do not trust it to skip ahead.
    TooLarge,
    /// The CRC does not match.
    Corrupt,
}

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct Frame<'a> {
    pub seq: u64,
    pub payload: &'a [u8],
}

/// CRC-32C (Castagnoli), bit by bit. crc32c(b"123456789") == 0xE306_9283.
/// (Production code uses the SSE4.2 / ARMv8 crc32c instructions.)
pub fn crc32c(bytes: &[u8]) -> u32 {
    let mut crc = !0u32;
    for &b in bytes {
        crc ^= b as u32;
        for _ in 0..8 {
            crc = if crc & 1 != 0 { (crc >> 1) ^ 0x82F6_3B78 } else { crc >> 1 };
        }
    }
    !crc
}

/// Append one frame for (`seq`, `payload`) to `out`.
pub fn encode_frame(seq: u64, payload: &[u8], out: &mut Vec<u8>) {
    todo!()
}

/// Decode the frame at the start of `buf`, borrowing its payload.
/// Returns the frame and the number of bytes it occupies. Check, in order:
///   fewer than HEADER bytes → Incomplete; len > MAX_PAYLOAD → TooLarge;
///   fewer than HEADER + len bytes → Incomplete; CRC mismatch → Corrupt.
pub fn decode_frame(buf: &[u8]) -> Result<(Frame<'_>, usize), FrameError> {
    todo!()
}

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Stop {
    Frame(FrameError),
    /// A frame decoded fine but its seq is not the previous seq + 1.
    Gap { expected: u64, got: u64 },
}

#[derive(Clone, PartialEq, Eq, Debug)]
pub struct Recovered {
    pub frames: Vec<(u64, Vec<u8>)>,
    /// Bytes of valid journal — truncate the file here before appending.
    pub valid_len: usize,
    /// Why the scan stopped early; None if it consumed the whole buffer.
    pub stopped: Option<Stop>,
}

/// Scan a journal from the start: keep decoding while frames are valid and
/// their seqs are consecutive (the first frame may start at any seq).
pub fn recover(buf: &[u8]) -> Recovered {
    todo!()
}

Week 7 listed the leaks inside the matching loop: clocks, HashMap iteration, floats, addresses, threads. Replay adds a few more, all of the same kind — something that influenced the state but is not in the journal:

• Configuration and reference data. Tick sizes, bands and limits loaded from a file at start-up are inputs. If they change, the change must go through the sequenced stream, or yesterday's journal replayed with today's config produces different rejects.
• Session events. Market open, close, halts and end-of-day purges must be sequenced commands, not timers inside the engine.
• Code version. A journal replayed through a newer engine whose rules changed will diverge, correctly. Record the engine version in the journal header and keep old behaviour reachable (or snapshot at upgrade boundaries) if you need old journals to replay exactly.
• Floating point outside prices. Even if prices are integers, a risk calculation using f64 transcendental functions can differ between platforms and library versions. Keep floats out of the replicated state, or make sure every replica runs the identical binary.

"Replay produces the same state" is only useful if you can check it cheaply. The tool is a state hash: serialize the state into a canonical byte form — one that depends only on the logical content, never on memory layout or iteration order — and hash it. Equal states give equal hashes on any machine; any difference in content changes the hash with overwhelming probability.

The trap is the one week 7 warned about: a HashMap's iteration order depends on its random seed, so two maps with identical contents iterate differently, and a "hash of the state" built by iterating one gives a different answer every run. Sort the keys (or keep the state in ordered structures) before serializing. The same canonical bytes double as a snapshot: restore from them, replay only the commands after the snapshot's sequence number, and you must land on the same hash as a full replay. Week 18 builds fast restarts on exactly that.

For a consistency check between replicas, a fast non-cryptographic hash such as FNV-1a or xxHash is enough; nobody is trying to forge a collision between your primary and your standby.

Crash the writer mid-frame and step through the restart: frames the snapshot already covers are only verified, the tail after it is replayed, and the recovered state hash must equal the uninterrupted run's. Switch the frame layout to the CRC-first format from Day 3's drill, then set the recovery code to "Off by one" to see why the applied count belongs in the hash.

The explainer's two layouts are the drills' formats (week 9's CRC first, week 18's CRC last). The repo's frame keeps the CRC in the header, after len and seq (see the capstone). The scan works the same way on all three.

use std::collections::HashMap;

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Side {
    Buy,
    Sell,
}

/// Inbound commands, exactly as the sequencer journaled them.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Cmd {
    Add { id: u64, side: Side, price: i64, qty: u64 },
    Reduce { id: u64, qty: u64 },
    Cancel { id: u64 },
}

#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct Resting {
    pub side: Side,
    pub price: i64,
    pub qty: u64,
}

/// Engine state: resting orders plus the number of commands applied.
/// Note the HashMap: two equal states can iterate in different orders.
#[derive(Clone, PartialEq, Eq, Debug)]
pub struct State {
    pub applied: u64,
    pub orders: HashMap<u64, Resting>,
}

/// 64-bit FNV-1a: start at 0xcbf29ce484222325; for each byte, XOR it in,
/// then multiply by 0x100000001b3 (wrapping).
/// fnv1a64(b"") == 0xcbf29ce484222325, fnv1a64(b"a") == 0xaf63dc4c8601ec8c.
pub fn fnv1a64(bytes: &[u8]) -> u64 {
    todo!()
}

impl State {
    pub fn new() -> Self {
        State { applied: 0, orders: HashMap::new() }
    }

    /// Apply one command. Every command — even a no-op — increments `applied`.
    ///   Add:    insert, unless the id already exists (then no-op)
    ///   Reduce: subtract min(qty, remaining); remove the order at zero;
    ///           unknown id → no-op
    ///   Cancel: remove; unknown id → no-op
    pub fn apply(&mut self, cmd: &Cmd) {
        todo!()
    }

    /// Canonical encoding — equal states give equal bytes, whatever order the
    /// HashMap iterates in. All integers little-endian:
    ///   applied u64 | count u64 | for each order by ascending id:
    ///   id u64 | side u8 (0 = Buy, 1 = Sell) | price i64 | qty u64
    pub fn canonical_bytes(&self) -> Vec<u8> {
        todo!()
    }

    pub fn state_hash(&self) -> u64 {
        fnv1a64(&self.canonical_bytes())
    }

    /// Inverse of canonical_bytes (this is your snapshot format). None if the
    /// bytes are truncated, have trailing bytes, a bad side byte, or ids that
    /// are not strictly ascending.
    pub fn restore(bytes: &[u8]) -> Option<State> {
        todo!()
    }
}

pub fn replay(cmds: &[Cmd]) -> State {
    let mut s = State::new();
    for c in cmds {
        s.apply(c);
    }
    s
}

A journal format is only half the design; the other half is what operators and code do when something is wrong. Decide in advance:

• Torn tail after a crash: truncate to the last valid frame and continue. This is normal.
• Corrupt frame or gap in the middle: stop. Do not skip it — every later state depends on it. Recover the missing range from a replica's journal, verify the hashes, and only then continue.
• Primary and standby hashes disagree: one of them is wrong, and you do not know which. Halt trading on the affected instruments rather than guessing; then replay both journals offline to find the first diverging command.

Emit a state hash periodically (every N commands, or at session events) into the output stream, so that divergence is detected minutes after it happens rather than at the end of the day.