Live, Stale, and Syncing

What you do: an explicit state machine

When a check fires, the program must stop trusting the top, stop the strategy acting on it, keep the deltas still arriving, fetch a snapshot, join the two, and resume. That is held as three named states, kept beside the book.

Live is the normal state: each delta is checked and applied, and the top is served to the strategy. Stale is entered the moment a gap, a cross, a bad quantity or a reconnect is seen. The top is not to be trusted; the strategy cancels the orders it has waiting in the venue’s book, or at least stops placing new ones; and deltas are buffered, not applied. Syncing begins when the snapshot has been requested, and deltas keep buffering. When the snapshot arrives, the handler loads it, uses the join rule to find the first buffered delta that continues it, replays from there, drops the deltas the snapshot already contains, and goes Live. Stale and Syncing are separate because the request cannot always go out at once (after a reconnect the stream must be reopened and buffering first, and venues limit how often a client may ask), and because the split tells an operator which half is slow.

Why a state and not an exception? The strategy must know, on every update, whether the number it reads can be trusted, and a state is something it can read: one more compare. An exception unwinds the thread to a handler that knows nothing of the orders waiting at the venue, and meanwhile those orders stay in the book and the market keeps moving. A gap is a normal event with its own path through the code, tested by breaking the feed on purpose.

The buffer needs a bound. At 20,000 updates a second, a 200 ms resync buffers 4,000 updates, 192 KB at the 48 bytes of the wire message. The buffer is allocated once, at startup, with room for 4,096, so that going Stale never asks the allocator for memory; if it fills, the resync has taken too long and starts again.

pub enum BookState { Live, Stale, Syncing { since_seq: u64 } }

// The feed handler, grown for recovery. Its fields: book: B, state: BookState,
// next_seq: u64, buffer: Vec<Update> with capacity 4,096 reserved at startup.
impl<B: Book> FeedHandler<B> {
    /// Returns the new top when it changed and can be trusted.
    pub fn on_update(&mut self, u: Update) -> Option<Top> {
        if let BookState::Live = self.state {
            if u.seq < self.next_seq { return None; }    // a repeat: already applied
            if u.seq == self.next_seq && u.qty >= 0 {
                self.next_seq = u.seq + 1;
                let changed = self.book.apply(&u);
                if !crossed(&self.book) { return changed.then(|| self.book.top()); }
            }
            self.state = BookState::Stale;               // gap, bad quantity or cross
            self.buffer.clear();
        }
        if self.buffer.len() == self.buffer.capacity() { // resync too slow: start again
            self.state = BookState::Stale;
            self.buffer.clear();
        }
        self.buffer.push(u);
        None
    }
}

impl<B: Book + Default> FeedHandler<B> {
    pub fn on_snapshot(&mut self, snapshot_id: u64, levels: &[Update]) -> Option<Top> {
        let BookState::Syncing { since_seq } = self.state else { return None };
        self.state = BookState::Stale;                   // until the join succeeds
        if since_seq > snapshot_id + 1 { return None; }  // snapshot older than the buffer
        self.book = B::default();                        // empty, then the snapshot
        for l in levels { self.book.apply(l); }
        self.next_seq = snapshot_id + 1;
        for u in self.buffer.drain(..) {
            if u.seq < self.next_seq { continue; }       // in the snapshot, or a repeat: drop
            let first = self.next_seq == snapshot_id + 1;
            if (first && !joins(snapshot_id, &u)) || (!first && u.seq != self.next_seq) {
                return None;                             // a hole: ask again
            }
            self.book.apply(&u);
            self.next_seq = u.seq + 1;
        }
        if crossed(&self.book) { return None; }
        self.state = BookState::Live;
        Some(self.book.top())
    }
}

crossed compares the two best prices; joins is the venue’s join rule, written from its documentation and not reimplemented here; B::default() stands for an empty book (a real handler clears the one it has). The caller sees Stale, tells the strategy, sends the request, and sets Syncing { since_seq } to the first buffered update’s number. The time spent in Stale and in Syncing is measured and alerted on, because it is the time the program is blind: no price it can trust, none of its orders in the market.