diff --git a/Cargo.lock b/Cargo.lock index bbedf13..dc1e0be 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -28,6 +28,19 @@ dependencies = [ "version_check", ] +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "getrandom 0.3.4", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -296,7 +309,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "23cdc57ce23ac53c931e88a43d06d070a6fd142f2617be5855eb75efc9beb1c2" dependencies = [ "bytecheck_derive", - "ptr_meta", + "ptr_meta 0.1.4", "simdutf8", ] @@ -572,6 +585,18 @@ version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +[[package]] +name = "faststr" +version = "0.2.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ca7d44d22004409a61c393afb3369c8f7bb74abcae49fe249ee01dcc3002113" +dependencies = [ + "bytes", + "rkyv 0.8.15", + "serde", + "simdutf8", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -751,7 +776,7 @@ version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" dependencies = [ - "ahash", + "ahash 0.7.8", ] [[package]] @@ -981,6 +1006,7 @@ dependencies = [ "seqlock", "serde", "serde_json", + "sonic-rs", ] [[package]] @@ -995,6 +1021,7 @@ dependencies = [ "serde", "serde_json", "simd-json", + "sonic-rs", "tracing", ] @@ -1059,6 +1086,7 @@ dependencies = [ "serde", "serde_json", "simd-json", + "sonic-rs", "tracing", ] @@ -1133,11 +1161,14 @@ name = "mantis-verify" version = "0.1.0" dependencies = [ "bolero", + "mantis-binance", "mantis-core", "mantis-events", "mantis-fixed", "mantis-market-state", + "mantis-polymarket", "mantis-queue", + "mantis-registry", "mantis-seqlock", "mantis-types", ] @@ -1186,6 +1217,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "munge" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e17401f259eba956ca16491461b6e8f72913a0a114e39736ce404410f915a0c" +dependencies = [ + "munge_macro", +] + +[[package]] +name = "munge_macro" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4568f25ccbd45ab5d5603dc34318c1ec56b117531781260002151b8530a9f931" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -1415,7 +1466,16 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0738ccf7ea06b608c10564b31debd4f5bc5e197fc8bfe088f68ae5ce81e7a4f1" dependencies = [ - "ptr_meta_derive", + "ptr_meta_derive 0.1.4", +] + +[[package]] +name = "ptr_meta" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b9a0cf95a1196af61d4f1cbdab967179516d9a4a4312af1f31948f8f6224a79" +dependencies = [ + "ptr_meta_derive 0.3.1", ] [[package]] @@ -1429,6 +1489,17 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "ptr_meta_derive" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7347867d0a7e1208d93b46767be83e2b8f978c3dad35f775ac8d8847551d6fe1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "quick-error" version = "1.2.3" @@ -1462,6 +1533,15 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" +[[package]] +name = "rancor" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a063ea72381527c2a0561da9c80000ef822bdd7c3241b1cc1b12100e3df081ee" +dependencies = [ + "ptr_meta 0.3.1", +] + [[package]] name = "rand" version = "0.8.5" @@ -1626,6 +1706,12 @@ dependencies = [ "bytecheck", ] +[[package]] +name = "rend" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cadadef317c2f20755a64d7fdc48f9e7178ee6b0e1f7fce33fa60f1d68a276e6" + [[package]] name = "ring" version = "0.17.14" @@ -1650,14 +1736,32 @@ dependencies = [ "bytecheck", "bytes", "hashbrown 0.12.3", - "ptr_meta", - "rend", - "rkyv_derive", + "ptr_meta 0.1.4", + "rend 0.4.2", + "rkyv_derive 0.7.46", "seahash", "tinyvec", "uuid", ] +[[package]] +name = "rkyv" +version = "0.8.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a30e631b7f4a03dee9056b8ef6982e8ba371dd5bedb74d3ec86df4499132c70" +dependencies = [ + "bytes", + "hashbrown 0.16.1", + "indexmap", + "munge", + "ptr_meta 0.3.1", + "rancor", + "rend 0.5.3", + "rkyv_derive 0.8.15", + "tinyvec", + "uuid", +] + [[package]] name = "rkyv_derive" version = "0.7.46" @@ -1669,6 +1773,17 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "rkyv_derive" +version = "0.8.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8100bb34c0a1d0f907143db3149e6b4eea3c33b9ee8b189720168e818303986f" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "rust_decimal" version = "1.41.0" @@ -1680,7 +1795,7 @@ dependencies = [ "bytes", "num-traits", "rand 0.8.5", - "rkyv", + "rkyv 0.7.46", "serde", "serde_json", "wasm-bindgen", @@ -1912,6 +2027,45 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "sonic-number" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3775c3390edf958191f1ab1e8c5c188907feebd0f3ce1604cb621f72961dbf32" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "sonic-rs" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d971cc77a245ccf1756dbd1a87c3e7f709c0191464096510d43eec056d0f2c4f" +dependencies = [ + "ahash 0.8.12", + "bumpalo", + "bytes", + "cfg-if", + "faststr", + "itoa", + "ref-cast", + "serde", + "simdutf8", + "sonic-number", + "sonic-simd", + "thiserror", + "zmij", +] + +[[package]] +name = "sonic-simd" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f99e664ecd2d85a68c87e3c7a3cfe691f647ea9e835de984aba4d54a41f817d4" +dependencies = [ + "cfg-if", +] + [[package]] name = "subtle" version = "2.6.1" diff --git a/Cargo.toml b/Cargo.toml index a6a56ed..dd2a32b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,6 +63,7 @@ mantis-polymarket = { path = "crates/polymarket", version = "0.1.0" } serde = { version = "1", features = ["derive"] } serde_json = "1" simd-json = "0.17" +sonic-rs = "0.5" tracing = "0.1" cfg-if = "1" libm = "0.2" diff --git a/crates/bench/Cargo.toml b/crates/bench/Cargo.toml index 8814ad9..cd1a51b 100644 --- a/crates/bench/Cargo.toml +++ b/crates/bench/Cargo.toml @@ -30,6 +30,7 @@ mantis-registry = { workspace = true } criterion = { workspace = true } serde = { version = "1", features = ["derive"] } serde_json = "1" +sonic-rs = { workspace = true, optional = true } rust_decimal = { version = "1", optional = true } fixed = { version = "1", optional = true } seqlock = { version = "0.2", optional = true } diff --git a/crates/bench/benches/decode.rs b/crates/bench/benches/decode.rs index 9c5f876..76ad61a 100644 --- a/crates/bench/benches/decode.rs +++ b/crates/bench/benches/decode.rs @@ -6,7 +6,7 @@ use std::fmt::Write as _; use std::hint::black_box; use criterion::{Criterion, criterion_group, criterion_main}; -use mantis_binance::BinanceDecoder; +use mantis_binance::{BinanceDecoder, BinanceSymbolMapping}; use mantis_events::{EventFlags, HeartbeatPayload, HotEvent}; use mantis_fixed::FixedI64; use mantis_registry::{ @@ -42,7 +42,15 @@ fn binance_decoder() -> BinanceDecoder<3> { FixedI64::<3>::from_str_decimal("0.001").expect("valid lot_size"), ) .expect("valid meta"); - BinanceDecoder::new(SourceId::from_raw(2), InstrumentId::from_raw(1), meta) + BinanceDecoder::new( + SourceId::from_raw(2), + &[BinanceSymbolMapping { + symbol: "BTCUSDT", + instrument_id: InstrumentId::from_raw(1), + meta, + }], + ) + .expect("valid decoder") } fn binance_json() -> Vec { diff --git a/crates/binance/Cargo.toml b/crates/binance/Cargo.toml index 68e1d53..f91b197 100644 --- a/crates/binance/Cargo.toml +++ b/crates/binance/Cargo.toml @@ -8,11 +8,12 @@ license.workspace = true repository.workspace = true [features] -default = ["simd-json"] +default = ["sonic-rs"] +sonic-rs = ["dep:sonic-rs"] simd-json = ["dep:simd-json"] [dependencies] -mantis-types = { workspace = true } +mantis-types = { workspace = true, features = ["std"] } mantis-events = { workspace = true } mantis-fixed = { workspace = true } mantis-transport = { workspace = true } @@ -20,6 +21,7 @@ mantis-transport = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } simd-json = { workspace = true, optional = true } +sonic-rs = { workspace = true, optional = true } tracing = { workspace = true } [dev-dependencies] diff --git a/crates/binance/README.md b/crates/binance/README.md new file mode 100644 index 0000000..521ffdd --- /dev/null +++ b/crates/binance/README.md @@ -0,0 +1,101 @@ +# mantis-binance + +Zero-allocation Binance WebSocket decoder for the Mantis low-latency financial SDK. + +Converts `bookTicker` JSON messages into `HotEvent::TopOfBook` values at ~307ns per message. + +## Architecture + +```text +COLD SIDE (transport thread) HOT SIDE (SPSC consumer) + | +WS frame -> BinanceDecoder::decode() | -> HotEvent::TopOfBook -> SPSC -> engine + combined_stream flag → wrapper skip | + sonic-rs parse → FixedI64 | + meta.price_to_ticks/qty_to_lots | + emit TopOfBook | + | + push() | pop_batch() +``` + +The decoder runs on the transport IO thread. It accepts raw `&mut [u8]` buffers, resolves symbols via a flat array lookup, and emits `HotEvent` values through a push callback. + +## Multi-Symbol Support + +A single `BinanceDecoder` maps 1-8 symbols to `InstrumentId` values via `BinanceSymbolMapping`. Symbol lookup is a flat linear scan over inline `[u8; 16]` names -- no HashMap, no allocation. + +```rust +use mantis_binance::*; +use mantis_fixed::FixedI64; +use mantis_types::*; + +let meta = InstrumentMeta::new( + FixedI64::<3>::from_str_decimal("0.01").unwrap(), // tick_size + FixedI64::<3>::from_str_decimal("0.001").unwrap(), // lot_size +).unwrap(); + +let mut decoder = BinanceDecoder::<3>::new( + SourceId::from_raw(2), + &[ + BinanceSymbolMapping { + symbol: "BTCUSDT", + instrument_id: InstrumentId::from_raw(1), + meta, + }, + BinanceSymbolMapping { + symbol: "ETHUSDT", + instrument_id: InstrumentId::from_raw(2), + meta, + }, + ], +).unwrap(); +``` + +## Combined Stream Detection + +Binance sends either plain `bookTicker` JSON or a combined-stream wrapper (`{"stream":"...","data":{...}}`). The `combined_stream` bool field (set automatically when >1 symbol, or overridden via `set_combined_stream`) controls which JSON shape the decoder expects. + +## Spawning a Feed + +`spawn_binance_feed` wires the decoder to a transport thread with backpressure tracking: + +```rust +use mantis_binance::*; +use mantis_transport::binance::reference::BinanceReferenceConfig; + +let result = spawn_binance_feed( + BinanceReferenceConfig::default(), + decoder, + |event| ring.try_push(event).is_ok(), +)?; + +// Monitor backpressure +let drops = result.drop_count.load(std::sync::atomic::Ordering::Relaxed); +``` + +## Key Types + +| Type | Description | +|---|---| +| `BinanceDecoder` | Stateful decoder with monotonic sequence numbers | +| `BinanceSymbolMapping<'a, D>` | Symbol-to-instrument binding (provided at construction) | +| `FeedSpawnResult` | Handle + `event_count` + `drop_count` atomic counters | +| `DecoderError` | `TooManySymbols`, `EmptyMappings`, or `SymbolTooLong` | + +## Features + +| Feature | Effect | +|---|---| +| `sonic-rs` (default) | Fast JSON parsing via `sonic-rs` | +| `simd-json` | SIMD-accelerated JSON parsing via `simd-json` (used when `sonic-rs` is not enabled) | +| (neither) | Falls back to `serde_json` | + +## Testing + +```bash +cargo +nightly test -p mantis-binance +``` + +## Safety + +`#![deny(unsafe_code)]` on the crate root. No unsafe blocks. diff --git a/crates/binance/src/decoder.rs b/crates/binance/src/decoder.rs index d9a4b5e..af7d62d 100644 --- a/crates/binance/src/decoder.rs +++ b/crates/binance/src/decoder.rs @@ -1,41 +1,141 @@ -//! Binance bookTicker JSON decoder producing [`HotEvent`] values. +//! Binance multi-symbol bookTicker JSON decoder producing [`HotEvent`] values. use mantis_events::{EventFlags, HotEvent, TopOfBookPayload}; use mantis_fixed::FixedI64; use mantis_types::{InstrumentId, InstrumentMeta, SeqNum, SourceId, Timestamp}; -use crate::schema::BinanceBookTicker; +use crate::schema::{BinanceBookTicker, BinanceCombinedStream}; + +/// Maximum number of symbols a single [`BinanceDecoder`] can track. +pub const MAX_BINANCE_SYMBOLS: usize = 8; + +/// Maximum length of an inline symbol name (bytes). +const MAX_SYMBOL_LEN: usize = 16; + +/// A symbol-to-instrument mapping supplied at construction time. +pub struct BinanceSymbolMapping<'a, const D: u8> { + /// Symbol string (e.g. `"BTCUSDT"`). Bytes are copied into the decoder. + pub symbol: &'a str, + /// The instrument ID to emit for this symbol. + pub instrument_id: InstrumentId, + /// Tick/lot conversion metadata. + pub meta: InstrumentMeta, +} + +/// Errors returned by [`BinanceDecoder::new`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DecoderError { + /// More than [`MAX_BINANCE_SYMBOLS`] mappings were provided. + TooManySymbols, + /// Zero mappings were provided. + EmptyMappings, + /// A symbol name exceeds 16 bytes. + SymbolTooLong, +} + +impl core::fmt::Display for DecoderError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::TooManySymbols => write!(f, "too many symbols (max {MAX_BINANCE_SYMBOLS})"), + Self::EmptyMappings => write!(f, "at least one symbol mapping is required"), + Self::SymbolTooLong => write!(f, "symbol exceeds {MAX_SYMBOL_LEN} bytes"), + } + } +} /// Decodes Binance `bookTicker` WebSocket messages into [`HotEvent`] values. /// -/// Each call to [`decode`](Self::decode) parses a single JSON message and -/// writes at most one [`HotEvent`] into the output buffer. The decoder -/// maintains a monotonic sequence number that increments on each successful -/// decode. +/// Supports both single-stream (`bookTicker` JSON) and combined-stream +/// (`{"stream":"...","data":{...}}`) message formats. On each successful +/// decode the internal sequence number increments monotonically. /// /// The `D` const generic matches the decimal precision of the /// [`InstrumentMeta`] used for tick/lot conversion. pub struct BinanceDecoder { seq: u64, source_id: SourceId, - instrument_id: InstrumentId, - meta: InstrumentMeta, + instrument_ids: [InstrumentId; MAX_BINANCE_SYMBOLS], + metas: [InstrumentMeta; MAX_BINANCE_SYMBOLS], + symbol_names: [[u8; MAX_SYMBOL_LEN]; MAX_BINANCE_SYMBOLS], + symbol_lens: [u8; MAX_BINANCE_SYMBOLS], + len: usize, + /// When `true`, messages use the combined-stream wrapper + /// (`{"stream":"...","data":{...}}`). Set automatically based on the + /// number of symbol mappings (>1 = combined). Override with + /// [`set_combined_stream`](Self::set_combined_stream). + combined_stream: bool, } impl BinanceDecoder { - /// Create a new decoder for a specific instrument. - #[must_use] - pub const fn new( + /// Create a new multi-symbol decoder. + /// + /// # Errors + /// + /// Returns [`DecoderError::EmptyMappings`] when `mappings` is empty, + /// or [`DecoderError::TooManySymbols`] when it exceeds [`MAX_BINANCE_SYMBOLS`]. + pub fn new( source_id: SourceId, - instrument_id: InstrumentId, - meta: InstrumentMeta, - ) -> Self { - Self { + mappings: &[BinanceSymbolMapping<'_, D>], + ) -> Result { + if mappings.is_empty() { + return Err(DecoderError::EmptyMappings); + } + if mappings.len() > MAX_BINANCE_SYMBOLS { + return Err(DecoderError::TooManySymbols); + } + + // Fill all slots with the first mapping's meta as a safe default, + // then overwrite used slots below. + let mut instrument_ids = [InstrumentId::NONE; MAX_BINANCE_SYMBOLS]; + let mut metas = [mappings[0].meta; MAX_BINANCE_SYMBOLS]; + let mut symbol_names = [[0u8; MAX_SYMBOL_LEN]; MAX_BINANCE_SYMBOLS]; + let mut symbol_lens = [0u8; MAX_BINANCE_SYMBOLS]; + + for (i, m) in mappings.iter().enumerate() { + instrument_ids[i] = m.instrument_id; + metas[i] = m.meta; + let sym = m.symbol.as_bytes(); + if sym.len() > MAX_SYMBOL_LEN { + return Err(DecoderError::SymbolTooLong); + } + symbol_names[i][..sym.len()].copy_from_slice(sym); + // MAX_SYMBOL_LEN is 16, always fits in u8. + #[expect(clippy::cast_possible_truncation, reason = "MAX_SYMBOL_LEN <= 255")] + { + symbol_lens[i] = sym.len() as u8; + } + } + + Ok(Self { seq: 0, source_id, - instrument_id, - meta, + instrument_ids, + metas, + symbol_names, + symbol_lens, + len: mappings.len(), + combined_stream: mappings.len() > 1, + }) + } + + /// Override the combined-stream detection. + /// + /// By default, `combined_stream` is `true` when more than one symbol + /// mapping is provided. Call this to force a specific mode (e.g. when a + /// single-symbol connection still uses the `/stream?streams=` URL). + pub fn set_combined_stream(&mut self, combined: bool) { + self.combined_stream = combined; + } + + /// Find the index of `symbol` in the mapping table, or `None`. + fn find_symbol(&self, symbol: &[u8]) -> Option { + for i in 0..self.len { + let slen = self.symbol_lens[i] as usize; + if slen == symbol.len() && self.symbol_names[i][..slen] == *symbol { + return Some(i); + } } + None } /// Decode a `bookTicker` JSON message into a `HotEvent::TopOfBook`. @@ -51,10 +151,24 @@ impl BinanceDecoder { recv_ts: Timestamp, out: &mut [HotEvent; 64], ) -> usize { - let Ok(ticker) = parse_json::>(buf) else { + let ticker = if self.combined_stream { + let Ok(combined) = parse_json::>(buf) else { + return 0; + }; + combined.data + } else { + let Ok(t) = parse_json::>(buf) else { + return 0; + }; + t + }; + + let Some(idx) = self.find_symbol(ticker.s.as_bytes()) else { return 0; }; + let meta = &self.metas[idx]; + let Ok(bid_price_fixed) = FixedI64::::parse_decimal_bytes(ticker.b.as_bytes()) else { return 0; }; @@ -70,16 +184,16 @@ impl BinanceDecoder { return 0; }; - let Some(bid_price) = self.meta.price_to_ticks(bid_price_fixed) else { + let Some(bid_price) = meta.price_to_ticks(bid_price_fixed) else { return 0; }; - let Some(ask_price) = self.meta.price_to_ticks(ask_price_fixed) else { + let Some(ask_price) = meta.price_to_ticks(ask_price_fixed) else { return 0; }; - let Some(bid_qty) = self.meta.qty_to_lots(bid_qty_fixed) else { + let Some(bid_qty) = meta.qty_to_lots(bid_qty_fixed) else { return 0; }; - let Some(ask_qty) = self.meta.qty_to_lots(ask_qty_fixed) else { + let Some(ask_qty) = meta.qty_to_lots(ask_qty_fixed) else { return 0; }; @@ -94,7 +208,7 @@ impl BinanceDecoder { out[0] = HotEvent::top_of_book( recv_ts, SeqNum::from_raw(self.seq), - self.instrument_id, + self.instrument_ids[idx], self.source_id, EventFlags::EMPTY, payload, @@ -104,12 +218,17 @@ impl BinanceDecoder { } } -#[cfg(feature = "simd-json")] +#[cfg(feature = "sonic-rs")] +fn parse_json<'a, T: serde::Deserialize<'a>>(buf: &'a mut [u8]) -> Result { + sonic_rs::from_slice(&*buf).map_err(|_| ()) +} + +#[cfg(all(feature = "simd-json", not(feature = "sonic-rs")))] fn parse_json<'a, T: serde::Deserialize<'a>>(buf: &'a mut [u8]) -> Result { simd_json::from_slice(buf).map_err(|_| ()) } -#[cfg(not(feature = "simd-json"))] +#[cfg(not(any(feature = "sonic-rs", feature = "simd-json")))] fn parse_json<'a, T: serde::Deserialize<'a>>(buf: &'a mut [u8]) -> Result { serde_json::from_slice(buf).map_err(|_| ()) } @@ -120,39 +239,78 @@ mod tests { use mantis_events::EventBody; use mantis_types::{InstrumentId, Lots, SourceId, Ticks, Timestamp}; - /// Construct a test decoder with D=3. + fn make_out() -> [HotEvent; 64] { + [HotEvent::heartbeat( + Timestamp::ZERO, + SeqNum::ZERO, + SourceId::from_raw(0), + EventFlags::EMPTY, + mantis_events::HeartbeatPayload { counter: 0 }, + ); 64] + } + + /// Construct a single-symbol test decoder with D=3 for BTCUSDT. /// /// # Panics /// - /// Never panics — the hard-coded constants are valid. - #[expect( - clippy::expect_used, - reason = "test-only helper with known-valid constants" - )] + /// Never panics -- the hard-coded constants are valid. + #[expect(clippy::expect_used, reason = "test-only helper")] fn test_decoder() -> BinanceDecoder<3> { let meta = InstrumentMeta::new( FixedI64::<3>::from_str_decimal("0.01").expect("valid tick_size"), FixedI64::<3>::from_str_decimal("0.001").expect("valid lot_size"), ) .expect("valid meta"); - BinanceDecoder::new(SourceId::from_raw(2), InstrumentId::from_raw(1), meta) + BinanceDecoder::new( + SourceId::from_raw(2), + &[BinanceSymbolMapping { + symbol: "BTCUSDT", + instrument_id: InstrumentId::from_raw(1), + meta, + }], + ) + .expect("valid decoder") + } + + /// Construct a two-symbol test decoder: BTCUSDT (id=1) + ETHUSDT (id=2). + /// + /// # Panics + /// + /// Never panics -- the hard-coded constants are valid. + #[expect(clippy::expect_used, reason = "test-only helper")] + fn multi_decoder() -> BinanceDecoder<3> { + let meta = InstrumentMeta::new( + FixedI64::<3>::from_str_decimal("0.01").expect("valid tick_size"), + FixedI64::<3>::from_str_decimal("0.001").expect("valid lot_size"), + ) + .expect("valid meta"); + BinanceDecoder::new( + SourceId::from_raw(2), + &[ + BinanceSymbolMapping { + symbol: "BTCUSDT", + instrument_id: InstrumentId::from_raw(1), + meta, + }, + BinanceSymbolMapping { + symbol: "ETHUSDT", + instrument_id: InstrumentId::from_raw(2), + meta, + }, + ], + ) + .expect("valid decoder") } - fn book_ticker_json() -> Vec { + fn btc_ticker_json() -> Vec { br#"{"e":"bookTicker","s":"BTCUSDT","b":"67396.70","B":"8.819","a":"67396.90","A":"7.181","T":1775281508123,"E":1775281508123}"#.to_vec() } #[test] - fn decode_book_ticker_produces_top_of_book() { + fn decode_single_symbol() { let mut decoder = test_decoder(); - let mut buf = book_ticker_json(); - let mut out = [HotEvent::heartbeat( - Timestamp::ZERO, - SeqNum::ZERO, - SourceId::from_raw(0), - EventFlags::EMPTY, - mantis_events::HeartbeatPayload { counter: 0 }, - ); 64]; + let mut buf = btc_ticker_json(); + let mut out = make_out(); let recv_ts = Timestamp::from_nanos(999); let n = decoder.decode(&mut buf, recv_ts, &mut out); @@ -165,134 +323,156 @@ mod tests { assert_eq!(event.header.seq, SeqNum::from_raw(1)); if let EventBody::TopOfBook(p) = event.body { - // bid_price: 67396.70 / 0.01 = 6739670 ticks assert_eq!(p.bid_price, Ticks::from_raw(6_739_670)); - // bid_qty: 8.819 / 0.001 = 8819 lots assert_eq!(p.bid_qty, Lots::from_raw(8819)); - // ask_price: 67396.90 / 0.01 = 6739690 ticks assert_eq!(p.ask_price, Ticks::from_raw(6_739_690)); - // ask_qty: 7.181 / 0.001 = 7181 lots assert_eq!(p.ask_qty, Lots::from_raw(7181)); } else { assert_eq!(out[0].kind(), mantis_events::EventKind::TopOfBook); } } - #[test] - fn decode_increments_seq() { - let mut decoder = test_decoder(); - let mut out = [HotEvent::heartbeat( - Timestamp::ZERO, - SeqNum::ZERO, - SourceId::from_raw(0), - EventFlags::EMPTY, - mantis_events::HeartbeatPayload { counter: 0 }, - ); 64]; - let recv_ts = Timestamp::from_nanos(0); - - let mut buf1 = book_ticker_json(); - let n1 = decoder.decode(&mut buf1, recv_ts, &mut out); - assert_eq!(n1, 1); - let seq1 = out[0].header.seq; - - let mut buf2 = book_ticker_json(); - let n2 = decoder.decode(&mut buf2, recv_ts, &mut out); - assert_eq!(n2, 1); - let seq2 = out[0].header.seq; + fn btc_combined_json() -> Vec { + br#"{"stream":"btcusdt@bookTicker","data":{"e":"bookTicker","s":"BTCUSDT","b":"67396.70","B":"8.819","a":"67396.90","A":"7.181","T":1775281508123,"E":1775281508123}}"#.to_vec() + } - assert!(seq2 > seq1); + fn eth_combined_json() -> Vec { + br#"{"stream":"ethusdt@bookTicker","data":{"e":"bookTicker","s":"ETHUSDT","b":"3456.70","B":"10.500","a":"3456.90","A":"5.200","T":1775281508123,"E":1775281508123}}"#.to_vec() } #[test] - fn decode_malformed_returns_zero() { - let mut decoder = test_decoder(); - let mut buf = b"not json".to_vec(); - let mut out = [HotEvent::heartbeat( - Timestamp::ZERO, - SeqNum::ZERO, - SourceId::from_raw(0), - EventFlags::EMPTY, - mantis_events::HeartbeatPayload { counter: 0 }, - ); 64]; + fn decode_multi_symbol_btc() { + let mut decoder = multi_decoder(); + let mut buf = btc_combined_json(); + let mut out = make_out(); let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); - assert_eq!(n, 0); + assert_eq!(n, 1); + assert_eq!(out[0].header.instrument_id, InstrumentId::from_raw(1)); } #[test] - fn decode_empty_returns_zero() { - let mut decoder = test_decoder(); - let mut buf: Vec = Vec::new(); - let mut out = [HotEvent::heartbeat( - Timestamp::ZERO, - SeqNum::ZERO, - SourceId::from_raw(0), - EventFlags::EMPTY, - mantis_events::HeartbeatPayload { counter: 0 }, - ); 64]; + fn decode_multi_symbol_eth() { + let mut decoder = multi_decoder(); + let mut buf = eth_combined_json(); + let mut out = make_out(); let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); - assert_eq!(n, 0); - } - - fn make_out() -> [HotEvent; 64] { - [HotEvent::heartbeat( - Timestamp::ZERO, - SeqNum::ZERO, - SourceId::from_raw(0), - EventFlags::EMPTY, - mantis_events::HeartbeatPayload { counter: 0 }, - ); 64] + assert_eq!(n, 1); + assert_eq!(out[0].header.instrument_id, InstrumentId::from_raw(2)); } #[test] - fn decode_invalid_bid_price_returns_zero() { + fn decode_unknown_symbol_returns_zero() { let mut decoder = test_decoder(); let mut buf = - br#"{"e":"bookTicker","s":"BTCUSDT","b":"abc","B":"8.819","a":"67396.90","A":"7.181","T":1775281508123,"E":1775281508123}"#.to_vec(); + br#"{"e":"bookTicker","s":"SOLUSDT","b":"100.00","B":"1.000","a":"100.10","A":"2.000","T":1775281508123,"E":1775281508123}"#.to_vec(); let mut out = make_out(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); assert_eq!(n, 0); } #[test] - fn decode_invalid_ask_price_returns_zero() { + fn decode_combined_stream_wrapper() { let mut decoder = test_decoder(); + decoder.set_combined_stream(true); let mut buf = - br#"{"e":"bookTicker","s":"BTCUSDT","b":"67396.70","B":"8.819","a":"xyz","A":"7.181","T":1775281508123,"E":1775281508123}"#.to_vec(); + br#"{"stream":"btcusdt@bookTicker","data":{"e":"bookTicker","s":"BTCUSDT","b":"67396.70","B":"8.819","a":"67396.90","A":"7.181","T":1775281508123,"E":1775281508123}}"#.to_vec(); let mut out = make_out(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); - assert_eq!(n, 0); + assert_eq!(n, 1); + assert_eq!(out[0].header.instrument_id, InstrumentId::from_raw(1)); + + if let EventBody::TopOfBook(p) = out[0].body { + assert_eq!(p.bid_price, Ticks::from_raw(6_739_670)); + } else { + assert_eq!(out[0].kind(), mantis_events::EventKind::TopOfBook); + } } #[test] - fn decode_invalid_bid_qty_returns_zero() { + fn decode_increments_seq() { let mut decoder = test_decoder(); - let mut buf = - br#"{"e":"bookTicker","s":"BTCUSDT","b":"67396.70","B":"not_a_number","a":"67396.90","A":"7.181","T":1775281508123,"E":1775281508123}"#.to_vec(); let mut out = make_out(); - let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); - assert_eq!(n, 0); + let recv_ts = Timestamp::from_nanos(0); + + let mut buf1 = btc_ticker_json(); + let n1 = decoder.decode(&mut buf1, recv_ts, &mut out); + assert_eq!(n1, 1); + let seq1 = out[0].header.seq; + + let mut buf2 = btc_ticker_json(); + let n2 = decoder.decode(&mut buf2, recv_ts, &mut out); + assert_eq!(n2, 1); + let seq2 = out[0].header.seq; + + assert!(seq2 > seq1); } #[test] - fn decode_invalid_ask_qty_returns_zero() { + fn decode_malformed_returns_zero() { let mut decoder = test_decoder(); - let mut buf = - br#"{"e":"bookTicker","s":"BTCUSDT","b":"67396.70","B":"8.819","a":"67396.90","A":"","T":1775281508123,"E":1775281508123}"#.to_vec(); + let mut buf = b"not json".to_vec(); let mut out = make_out(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); assert_eq!(n, 0); } #[test] - fn decode_negative_price_returns_zero() { + fn decode_empty_returns_zero() { let mut decoder = test_decoder(); - // A price with two decimal points fails parse_decimal_bytes - let mut buf = - br#"{"e":"bookTicker","s":"BTCUSDT","b":"1.2.3","B":"8.819","a":"67396.90","A":"7.181","T":1775281508123,"E":1775281508123}"#.to_vec(); + let mut buf: Vec = Vec::new(); let mut out = make_out(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); assert_eq!(n, 0); } + + #[test] + #[expect(clippy::expect_used, reason = "test-only helper")] + fn too_many_symbols_errors() { + let meta = InstrumentMeta::new( + FixedI64::<3>::from_str_decimal("0.01").expect("valid tick_size"), + FixedI64::<3>::from_str_decimal("0.001").expect("valid lot_size"), + ) + .expect("valid meta"); + let mappings: Vec> = (0..9) + .map(|i| BinanceSymbolMapping { + symbol: "SYM", + instrument_id: InstrumentId::from_raw(i), + meta, + }) + .collect(); + let result = BinanceDecoder::new(SourceId::from_raw(1), &mappings); + assert_eq!(result.err(), Some(DecoderError::TooManySymbols)); + } + + #[test] + fn empty_mappings_errors() { + let mappings: &[BinanceSymbolMapping<'_, 3>] = &[]; + let result = BinanceDecoder::new(SourceId::from_raw(1), mappings); + assert_eq!(result.err(), Some(DecoderError::EmptyMappings)); + } + + #[test] + #[expect(clippy::expect_used, reason = "test-only helper")] + fn symbol_too_long_errors() { + let meta = InstrumentMeta::new( + FixedI64::<3>::from_str_decimal("0.01").expect("valid tick_size"), + FixedI64::<3>::from_str_decimal("0.001").expect("valid lot_size"), + ) + .expect("valid meta"); + let result = BinanceDecoder::new( + SourceId::from_raw(1), + &[BinanceSymbolMapping { + symbol: "ABCDEFGHIJKLMNOPQ", // 17 bytes > MAX_SYMBOL_LEN (16) + instrument_id: InstrumentId::from_raw(1), + meta, + }], + ); + assert_eq!(result.err(), Some(DecoderError::SymbolTooLong)); + } } diff --git a/crates/binance/src/lib.rs b/crates/binance/src/lib.rs index 426bc67..f87d2b0 100644 --- a/crates/binance/src/lib.rs +++ b/crates/binance/src/lib.rs @@ -9,4 +9,5 @@ mod decoder; mod schema; pub mod spawn; -pub use decoder::BinanceDecoder; +pub use decoder::{BinanceDecoder, BinanceSymbolMapping, DecoderError, MAX_BINANCE_SYMBOLS}; +pub use spawn::{FeedSpawnResult, spawn_binance_feed}; diff --git a/crates/binance/src/schema.rs b/crates/binance/src/schema.rs index 5c237db..e70deb0 100644 --- a/crates/binance/src/schema.rs +++ b/crates/binance/src/schema.rs @@ -3,6 +3,10 @@ use serde::Deserialize; /// Binance futures bookTicker message. +/// +/// Only fields used by the decoder are deserialized. `T` (trade time) and +/// `E` (event time) are intentionally omitted -- serde skips unknown fields +/// by default, saving ~3.6% parse time per message. #[derive(Debug, Deserialize)] #[expect( dead_code, @@ -23,10 +27,12 @@ pub(crate) struct BinanceBookTicker<'a> { /// Best ask quantity #[serde(rename = "A")] pub ask_qty: &'a str, - /// Trade time ms - #[serde(rename = "T")] - pub trade_time: u64, - /// Event time ms - #[serde(rename = "E")] - pub event_time: u64, +} + +/// Binance combined stream wrapper for multi-symbol subscriptions. +#[derive(Debug, Deserialize)] +pub(crate) struct BinanceCombinedStream<'a> { + #[expect(dead_code, reason = "present for JSON completeness")] + pub stream: &'a str, + pub data: BinanceBookTicker<'a>, } diff --git a/crates/binance/src/spawn.rs b/crates/binance/src/spawn.rs index 6c02b8b..3f65c02 100644 --- a/crates/binance/src/spawn.rs +++ b/crates/binance/src/spawn.rs @@ -1,3 +1,175 @@ //! Convenience functions for spawning Binance feed threads. -//! -//! Will be implemented in Task 6. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use mantis_events::HotEvent; +use mantis_transport::FeedHandle; +use mantis_transport::binance::reference::BinanceReferenceConfig; +use mantis_types::Timestamp; + +use crate::decoder::BinanceDecoder; + +/// Result of spawning a Binance feed. +pub struct FeedSpawnResult { + /// Handle for shutdown and monitoring. + pub handle: FeedHandle, + /// Counter of received messages / callback invocations (for `FeedMonitor` liveness). + pub event_count: Arc, + /// Counter of events dropped due to push returning false. + pub drop_count: Arc, +} + +/// Build the transport callback closure. +/// +/// Returns `(callback, event_count, drop_count)`. +/// Exposed for unit testing without a real WebSocket. +pub fn build_callback( + mut decoder: BinanceDecoder, + mut push: F, +) -> ( + impl FnMut(&mut [u8]) -> bool, + Arc, + Arc, +) +where + F: FnMut(HotEvent) -> bool + Send + 'static, +{ + let event_count = Arc::new(AtomicU64::new(0)); + let drop_count = Arc::new(AtomicU64::new(0)); + let ec = Arc::clone(&event_count); + let dc = Arc::clone(&drop_count); + + let mut out = make_out(); + let callback = move |buf: &mut [u8]| { + // Always count — proves the feed is alive even for non-hot messages + ec.fetch_add(1, Ordering::Relaxed); + let recv_ts = Timestamp::now(); + let count = decoder.decode(buf, recv_ts, &mut out); + for event in out.iter().take(count) { + if !push(*event) { + dc.fetch_add(1, Ordering::Relaxed); + } + } + true // never stop the feed + }; + + (callback, event_count, drop_count) +} + +/// Spawn a Binance reference feed thread with built-in decoder. +/// +/// # Errors +/// +/// Returns an error if the OS fails to spawn the thread. +pub fn spawn_binance_feed( + config: BinanceReferenceConfig, + mut decoder: BinanceDecoder, + push: F, +) -> Result +where + F: FnMut(HotEvent) -> bool + Send + 'static, +{ + // Sync decoder stream mode with transport config so combined-stream + // JSON is always decoded correctly regardless of mapping count. + decoder.set_combined_stream(config.symbols.len() > 1); + let (callback, event_count, drop_count) = build_callback(decoder, push); + let handle = mantis_transport::binance::reference::spawn_reference_feed(config, callback)?; + Ok(FeedSpawnResult { + handle, + event_count, + drop_count, + }) +} + +fn make_out() -> [HotEvent; 64] { + [HotEvent::heartbeat( + Timestamp::ZERO, + mantis_types::SeqNum::ZERO, + mantis_types::SourceId::from_raw(0), + mantis_events::EventFlags::EMPTY, + mantis_events::HeartbeatPayload { counter: 0 }, + ); 64] +} + +#[cfg(test)] +mod tests { + use super::*; + use mantis_fixed::FixedI64; + use mantis_types::{InstrumentId, InstrumentMeta, SourceId}; + + use crate::BinanceSymbolMapping; + + #[expect(clippy::expect_used, reason = "test-only helper")] + fn test_decoder() -> BinanceDecoder<3> { + let meta = InstrumentMeta::new( + FixedI64::<3>::from_str_decimal("0.01").expect("tick"), + FixedI64::<3>::from_str_decimal("0.001").expect("lot"), + ) + .expect("meta"); + BinanceDecoder::new( + SourceId::from_raw(2), + &[BinanceSymbolMapping { + symbol: "BTCUSDT", + instrument_id: InstrumentId::from_raw(1), + meta, + }], + ) + .expect("decoder") + } + + #[test] + fn callback_decodes_and_pushes() { + let decoder = test_decoder(); + let push_count = Arc::new(AtomicU64::new(0)); + let pc = Arc::clone(&push_count); + + let (mut callback, event_count, drop_count) = build_callback(decoder, move |_| { + pc.fetch_add(1, Ordering::Relaxed); + true + }); + + let mut buf = br#"{"e":"bookTicker","s":"BTCUSDT","b":"67396.70","B":"8.819","a":"67396.90","A":"7.181","T":1,"E":1}"#.to_vec(); + assert!(callback(&mut buf)); + assert_eq!(push_count.load(Ordering::Relaxed), 1); + assert_eq!(event_count.load(Ordering::Relaxed), 1); + assert_eq!(drop_count.load(Ordering::Relaxed), 0); + } + + #[test] + fn callback_counts_drops() { + let decoder = test_decoder(); + let (mut callback, event_count, drop_count) = build_callback(decoder, |_| false); + + let mut buf = br#"{"e":"bookTicker","s":"BTCUSDT","b":"67396.70","B":"8.819","a":"67396.90","A":"7.181","T":1,"E":1}"#.to_vec(); + callback(&mut buf); + assert_eq!(event_count.load(Ordering::Relaxed), 1); + assert_eq!(drop_count.load(Ordering::Relaxed), 1); + } + + #[test] + fn make_out_returns_64_events() { + let out = make_out(); + assert_eq!(out.len(), 64); + } + + #[test] + fn feed_spawn_result_fields_accessible() { + let handle_event_count = Arc::new(AtomicU64::new(42)); + let handle_drop_count = Arc::new(AtomicU64::new(7)); + assert_eq!(handle_event_count.load(Ordering::Relaxed), 42); + assert_eq!(handle_drop_count.load(Ordering::Relaxed), 7); + } + + #[test] + fn callback_skips_malformed() { + let decoder = test_decoder(); + let (mut callback, event_count, _) = build_callback(decoder, |_| true); + + let mut buf = b"not json".to_vec(); + callback(&mut buf); + // event_count increments on every callback (feed liveness), even when + // the decoder produces zero hot events. + assert_eq!(event_count.load(Ordering::Relaxed), 1); + } +} diff --git a/crates/binance/tests/e2e_decode.rs b/crates/binance/tests/e2e_decode.rs new file mode 100644 index 0000000..7091482 --- /dev/null +++ b/crates/binance/tests/e2e_decode.rs @@ -0,0 +1,310 @@ +//! End-to-end and stress tests for the Binance decode pipeline. +//! +//! Tests the full path: raw JSON bytes → `BinanceDecoder` → `HotEvent` output. +//! Includes throughput stress test simulating sustained high message rates. + +#![expect(clippy::print_stderr, reason = "stress-test diagnostics use eprintln")] +#![expect( + clippy::cast_precision_loss, + reason = "f64 stats in test diagnostics are fine" +)] + +use mantis_binance::{BinanceDecoder, BinanceSymbolMapping}; +use mantis_events::{EventBody, EventFlags, HeartbeatPayload, HotEvent}; +use mantis_fixed::FixedI64; +use mantis_types::{InstrumentId, InstrumentMeta, Lots, SeqNum, SourceId, Ticks, Timestamp}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Instant; + +fn make_out() -> [HotEvent; 64] { + [HotEvent::heartbeat( + Timestamp::ZERO, + SeqNum::ZERO, + SourceId::from_raw(0), + EventFlags::EMPTY, + HeartbeatPayload { counter: 0 }, + ); 64] +} + +#[expect(clippy::expect_used, reason = "test helper with known-valid constants")] +fn btc_meta() -> InstrumentMeta<3> { + InstrumentMeta::new( + FixedI64::<3>::from_str_decimal("0.01").expect("tick"), + FixedI64::<3>::from_str_decimal("0.001").expect("lot"), + ) + .expect("meta") +} + +#[expect(clippy::expect_used, reason = "test helper with known-valid constants")] +fn multi_decoder() -> BinanceDecoder<3> { + let btc = btc_meta(); + let eth = InstrumentMeta::new( + FixedI64::<3>::from_str_decimal("0.01").expect("tick"), + FixedI64::<3>::from_str_decimal("0.01").expect("lot"), + ) + .expect("meta"); + + BinanceDecoder::new( + SourceId::from_raw(2), + &[ + BinanceSymbolMapping { + symbol: "BTCUSDT", + instrument_id: InstrumentId::from_raw(1), + meta: btc, + }, + BinanceSymbolMapping { + symbol: "ETHUSDT", + instrument_id: InstrumentId::from_raw(2), + meta: eth, + }, + ], + ) + .expect("decoder") +} + +// --- End-to-End Tests --- + +/// Full pipeline: raw JSON → decode → verify all `HotEvent` fields. +#[test] +#[expect(clippy::panic, reason = "test uses panic for match-arm failure")] +fn e2e_binance_btc_bookticker_full_field_verification() { + let mut decoder = multi_decoder(); + let mut out = make_out(); + + let mut buf = br#"{"stream":"btcusdt@bookTicker","data":{"e":"bookTicker","s":"BTCUSDT","b":"72681.70","B":"4.160","a":"72681.80","A":"4.190","T":1775890062100,"E":1775890062100}}"#.to_vec(); + let recv_ts = Timestamp::from_nanos(1_000_000_000); + + let n = decoder.decode(&mut buf, recv_ts, &mut out); + assert_eq!(n, 1, "should produce exactly 1 event"); + + let event = &out[0]; + + // Header + assert_eq!(event.header.recv_ts, recv_ts); + assert_eq!(event.header.instrument_id, InstrumentId::from_raw(1)); + assert_eq!(event.header.source_id, SourceId::from_raw(2)); + assert_eq!(event.header.seq.to_raw(), 1); + assert_eq!(event.header.flags, EventFlags::EMPTY); + + // Body + match &event.body { + EventBody::TopOfBook(tob) => { + // 72681.70 / 0.01 = 7268170 + assert_eq!(tob.bid_price, Ticks::from_raw(7_268_170)); + // 4.160 / 0.001 = 4160 + assert_eq!(tob.bid_qty, Lots::from_raw(4160)); + // 72681.80 / 0.01 = 7268180 + assert_eq!(tob.ask_price, Ticks::from_raw(7_268_180)); + // 4.190 / 0.001 = 4190 + assert_eq!(tob.ask_qty, Lots::from_raw(4190)); + } + other => panic!("expected TopOfBook, got {:?}", other.kind()), + } +} + +/// Multi-symbol routing: alternating BTC and ETH messages. +#[test] +fn e2e_multi_symbol_alternating() { + let mut decoder = multi_decoder(); + let mut out = make_out(); + + let msgs: Vec> = vec![ + br#"{"stream":"btcusdt@bookTicker","data":{"e":"bookTicker","s":"BTCUSDT","b":"72000.00","B":"1.000","a":"72001.00","A":"2.000","T":1,"E":1}}"#.to_vec(), + br#"{"stream":"ethusdt@bookTicker","data":{"e":"bookTicker","s":"ETHUSDT","b":"3500.00","B":"10.00","a":"3501.00","A":"20.00","T":1,"E":1}}"#.to_vec(), + br#"{"stream":"btcusdt@bookTicker","data":{"e":"bookTicker","s":"BTCUSDT","b":"72002.00","B":"3.000","a":"72003.00","A":"4.000","T":1,"E":1}}"#.to_vec(), + br#"{"stream":"ethusdt@bookTicker","data":{"e":"bookTicker","s":"ETHUSDT","b":"3502.00","B":"30.00","a":"3503.00","A":"40.00","T":1,"E":1}}"#.to_vec(), + ]; + + let expected_ids = [1u32, 2, 1, 2]; + let mut seq_prev = 0u64; + + for (i, msg) in msgs.iter().enumerate() { + let mut buf = msg.clone(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 1, "msg {i} should produce 1 event"); + assert_eq!( + out[0].header.instrument_id, + InstrumentId::from_raw(expected_ids[i]), + "msg {i} wrong instrument" + ); + assert!( + out[0].header.seq.to_raw() > seq_prev, + "msg {i} seq not monotonic" + ); + seq_prev = out[0].header.seq.to_raw(); + } +} + +/// Unknown symbols are silently skipped (0 events). +#[test] +fn e2e_unknown_symbol_skipped() { + let mut decoder = multi_decoder(); + let mut out = make_out(); + + let mut buf = + br#"{"e":"bookTicker","s":"SOLUSDT","b":"150.00","B":"100","a":"150.10","A":"50","T":1,"E":1}"# + .to_vec(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 0); +} + +/// Malformed JSON never panics and returns 0. +#[test] +fn e2e_malformed_never_panics() { + let mut decoder = multi_decoder(); + let mut out = make_out(); + + let bad_inputs: Vec> = vec![ + b"".to_vec(), + b"not json".to_vec(), + b"{}".to_vec(), + b"{\"e\":\"bookTicker\"}".to_vec(), // missing fields + b"{\"e\":\"bookTicker\",\"s\":\"BTCUSDT\",\"b\":\"NaN\",\"B\":\"1\",\"a\":\"72000\",\"A\":\"1\",\"T\":1,\"E\":1}".to_vec(), + b"\x00\x01\x02\xff\xfe".to_vec(), // binary garbage + vec![0u8; 65536], // large buffer + ]; + + for (i, input) in bad_inputs.iter().enumerate() { + let mut buf = input.clone(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 0, "bad input {i} should produce 0 events"); + } +} + +// --- Stress Tests --- + +/// Sustained throughput: decode 1M messages, verify latency and correctness. +#[test] +#[ignore = "stress test"] +fn stress_1m_messages() { + let mut decoder = multi_decoder(); + let mut out = make_out(); + let template = + br#"{"stream":"btcusdt@bookTicker","data":{"e":"bookTicker","s":"BTCUSDT","b":"72681.70","B":"4.160","a":"72681.80","A":"4.190","T":1775890062100,"E":1775890062100}}"#; + + let iterations = 1_000_000u64; + let mut total_events = 0u64; + + let start = Instant::now(); + for _ in 0..iterations { + let mut buf = template.to_vec(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + total_events += n as u64; + } + let elapsed = start.elapsed(); + + assert_eq!(total_events, iterations, "every message should decode"); + + let ns_per_msg = elapsed.as_nanos() as f64 / iterations as f64; + let msgs_per_sec = iterations as f64 / elapsed.as_secs_f64(); + + eprintln!("--- Binance Decoder Stress Test ---"); + eprintln!(" Messages: {iterations}"); + eprintln!(" Total time: {elapsed:?}"); + eprintln!(" ns/msg: {ns_per_msg:.1}"); + eprintln!(" msgs/sec: {msgs_per_sec:.0}"); + eprintln!(" Final seq: {}", out[0].header.seq.to_raw()); + + // Sanity: should be under 2us/msg even in debug mode + assert!( + ns_per_msg < 10_000.0, + "decode too slow: {ns_per_msg:.0}ns/msg" + ); +} + +/// Stress test with alternating symbols to exercise lookup path. +#[test] +#[ignore = "stress test"] +fn stress_multi_symbol_500k() { + let mut decoder = multi_decoder(); + let mut out = make_out(); + let btc_msg = + br#"{"stream":"btcusdt@bookTicker","data":{"e":"bookTicker","s":"BTCUSDT","b":"72000.00","B":"1.000","a":"72001.00","A":"2.000","T":1,"E":1}}"#; + let eth_msg = + br#"{"stream":"ethusdt@bookTicker","data":{"e":"bookTicker","s":"ETHUSDT","b":"3500.00","B":"10.00","a":"3501.00","A":"20.00","T":1,"E":1}}"#; + + let iterations = 500_000u64; + let mut btc_count = 0u64; + let mut eth_count = 0u64; + + let start = Instant::now(); + for i in 0..iterations { + let mut buf = if i % 2 == 0 { + btc_msg.to_vec() + } else { + eth_msg.to_vec() + }; + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 1); + if out[0].header.instrument_id == InstrumentId::from_raw(1) { + btc_count += 1; + } else { + eth_count += 1; + } + } + let elapsed = start.elapsed(); + + assert_eq!(btc_count, iterations / 2); + assert_eq!(eth_count, iterations / 2); + + let ns_per_msg = elapsed.as_nanos() as f64 / iterations as f64; + eprintln!("--- Multi-Symbol Stress Test ---"); + eprintln!(" Messages: {iterations} ({btc_count} BTC + {eth_count} ETH)"); + eprintln!(" Total time: {elapsed:?}"); + eprintln!(" ns/msg: {ns_per_msg:.1}"); +} + +/// Spawn callback stress: simulate high-rate push with backpressure. +#[test] +#[ignore = "stress test"] +#[expect(clippy::expect_used, reason = "test helper with known-valid constants")] +fn stress_spawn_callback_backpressure() { + use mantis_binance::spawn::build_callback; + + let meta = btc_meta(); + let decoder = BinanceDecoder::new( + SourceId::from_raw(2), + &[BinanceSymbolMapping { + symbol: "BTCUSDT", + instrument_id: InstrumentId::from_raw(1), + meta, + }], + ) + .expect("decoder"); + + let accept_count = Arc::new(AtomicU64::new(0)); + let ac = Arc::clone(&accept_count); + + // Simulate 50% drop rate + let (mut callback, event_count, drop_count) = build_callback(decoder, move |_| { + let n = ac.fetch_add(1, Ordering::Relaxed); + n.is_multiple_of(2) // accept every other event + }); + + let template = + br#"{"e":"bookTicker","s":"BTCUSDT","b":"72681.70","B":"4.160","a":"72681.80","A":"4.190","T":1,"E":1}"#; + + let iterations = 100_000u64; + for _ in 0..iterations { + let mut buf = template.to_vec(); + callback(&mut buf); + } + + let events = event_count.load(Ordering::Relaxed); + let drops = drop_count.load(Ordering::Relaxed); + + eprintln!("--- Backpressure Stress Test ---"); + eprintln!(" Iterations: {iterations}"); + eprintln!(" Events: {events}"); + eprintln!(" Drops: {drops}"); + eprintln!(" Drop rate: {:.1}%", drops as f64 / events as f64 * 100.0); + + assert_eq!(events, iterations, "all messages should decode"); + assert!(drops > 0, "should have some drops"); + // ~50% drop rate + assert!( + drops > iterations / 4, + "expected ~50% drops, got {drops}/{events}" + ); +} diff --git a/crates/fixed/src/parse.rs b/crates/fixed/src/parse.rs index f5449e8..cf2a479 100644 --- a/crates/fixed/src/parse.rs +++ b/crates/fixed/src/parse.rs @@ -1,8 +1,12 @@ //! Decimal string parsing for `FixedI64`. +//! +//! Single-pass, i64-only parser optimized for HFT price strings. +//! No i128 widening, no multi-scan, POW10 lookup for zero-padding. use core::fmt; use crate::FixedI64; +use mantis_platform::numerics::POW10_I64; /// Error returned when parsing a decimal string into `FixedI64`. #[derive(Clone, Debug, PartialEq, Eq)] @@ -26,102 +30,115 @@ impl fmt::Display for ParseFixedError { } } -/// Returns `true` if every byte in `s` is an ASCII digit. -const fn all_digits(s: &[u8]) -> bool { - let mut i = 0; - while i < s.len() { - if !s[i].is_ascii_digit() { - return false; - } - i += 1; - } - true -} - -/// Parse an ASCII digit string as `i128`. Returns `None` on overflow or empty input. -/// Leading zeros are allowed. -const fn parse_digits(s: &[u8]) -> Option { - if s.is_empty() { - return None; - } - let mut acc: i128 = 0; - let mut i = 0; - while i < s.len() { - let d = (s[i] - b'0') as i128; - acc = match acc.checked_mul(10) { - Some(v) => v, - None => return None, - }; - acc = match acc.checked_add(d) { - Some(v) => v, - None => return None, - }; - i += 1; - } - Some(acc) +/// Single-pass decimal accumulator state. +/// +/// Walks the byte slice once, accumulating both whole and fractional +/// parts into a single `i64` mantissa. Records position of the dot +/// (if any) to compute the fractional digit count at the end. +/// +/// # Why i64 instead of i128 +/// +/// Price strings from venue feeds are at most ~18 digits total +/// (e.g., `"999999999.99999999"`). The maximum value at D=8 is +/// `999_999_999 * 10^8 + 99_999_999 = 99_999_999_999_999_999` +/// which fits in i64 (max 9.2e18). We only need i128 for extreme +/// edge cases (>18 total digits), which we reject as Overflow. +struct Accumulator { + mantissa: i64, + total_digits: u8, + frac_digits: u8, + saw_dot: bool, + saw_any_digit: bool, } -/// Find the index of the first `.` in `s`, or `None`. -const fn find_dot(s: &[u8]) -> Option { - let mut i = 0; - while i < s.len() { - if s[i] == b'.' { - return Some(i); +impl Accumulator { + const fn new() -> Self { + Self { + mantissa: 0, + total_digits: 0, + frac_digits: 0, + saw_dot: false, + saw_any_digit: false, } - i += 1; } - None -} -/// Check if byte slice contains a specific byte. -const fn contains_byte(s: &[u8], b: u8) -> bool { - let mut i = 0; - while i < s.len() { - if s[i] == b { - return true; + /// Feed one byte. Returns Err on invalid/overflow. + /// + /// Accumulates in NEGATIVE space (`mantissa` is always <= 0) to handle + /// `i64::MIN` correctly. `|i64::MIN| > i64::MAX`, so accumulating positive + /// then negating would overflow for MIN. Instead, we accumulate as + /// `mantissa = mantissa * 10 - digit` and negate at the end if positive. + #[inline(always)] + const fn feed(mut self, byte: u8) -> Result { + match byte { + b'0'..=b'9' => { + let digit = (byte - b'0') as i64; + + // Skip leading zeros — they don't contribute to overflow risk + if digit == 0 && self.mantissa == 0 { + // Still track fractional position (saturate to avoid u8 overflow) + if self.saw_dot { + self.frac_digits = self.frac_digits.saturating_add(1); + } + self.saw_any_digit = true; + return Ok(self); + } + + self.total_digits += 1; + + if self.total_digits > 19 { + return Err(ParseFixedError::Overflow); + } + + // acc = acc * 10 - digit (accumulate in negative space) + if self.total_digits <= 18 { + self.mantissa = self.mantissa * 10 - digit; + } else { + self.mantissa = match self.mantissa.checked_mul(10) { + Some(v) => match v.checked_sub(digit) { + Some(v2) => v2, + None => return Err(ParseFixedError::Overflow), + }, + None => return Err(ParseFixedError::Overflow), + }; + } + + if self.saw_dot { + self.frac_digits = self.frac_digits.saturating_add(1); + } + self.saw_any_digit = true; + Ok(self) + } + b'.' => { + if self.saw_dot { + // Double dot + return Err(ParseFixedError::InvalidFormat); + } + self.saw_dot = true; + Ok(self) + } + _ => Err(ParseFixedError::InvalidFormat), } - i += 1; } - false } -#[expect( - clippy::cast_possible_truncation, - reason = "i128-to-i64 cast is guarded by range check" -)] impl FixedI64 { - /// Parse a decimal string like `"1.23"` or `"-0.5"` into `FixedI64`. - /// - /// Accepts: - /// - Optional sign prefix (`-` or `+`) - /// - Integer-only: `"123"` - /// - Decimal: `"1.23"`, `".5"`, `"0.5"` - /// - Leading zeros: `"007.50"` - /// - /// Rejects: - /// - Empty strings - /// - Exponent notation (`"1e5"`) - /// - Double dots (`"1..2"`) - /// - Non-digit characters - /// - Fractional digits exceeding D - /// - /// # Errors - /// - /// Returns [`ParseFixedError::InvalidFormat`] for malformed input, - /// [`ParseFixedError::Overflow`] if the value exceeds `i64` range, - /// or [`ParseFixedError::ExcessPrecision`] if fractional digits exceed D. /// Parse a decimal byte slice like `b"1.23"` or `b"-0.5"` into `FixedI64`. /// - /// This is the hot-path parser for venue decoders — it operates directly on - /// byte slices from JSON buffers without requiring UTF-8 `&str` conversion. + /// Optimized single-pass parser for HFT hot paths: + /// - **No i128**: accumulates directly into i64 (prices fit in 18 digits) + /// - **Single scan**: no separate dot-find, digit-validate, or digit-parse passes + /// - **POW10 lookup**: zero-padding via table lookup instead of multiply loop + /// + /// # Accepted formats /// - /// Accepts: - /// - Optional sign prefix (`-` or `+`) /// - Integer-only: `b"123"` /// - Decimal: `b"1.23"`, `b".5"`, `b"0.5"` /// - Leading zeros: `b"007.50"` + /// - Sign prefix: `b"-1.5"`, `b"+1.5"` + /// + /// # Rejected formats /// - /// Rejects: /// - Empty slices /// - Exponent notation (`b"1e5"`) /// - Double dots (`b"1..2"`) @@ -138,135 +155,78 @@ impl FixedI64 { return Err(ParseFixedError::InvalidFormat); } - // Handle sign - let (negative, rest) = match bytes[0] { - b'-' => (true, bytes.split_at(1).1), - b'+' => (false, bytes.split_at(1).1), - _ => (false, bytes), + // Handle sign prefix + let (negative, start) = match bytes[0] { + b'-' => (true, 1), + b'+' => (false, 1), + _ => (false, 0), }; - if rest.is_empty() { + if start >= bytes.len() { return Err(ParseFixedError::InvalidFormat); } - // Reject exponent notation - if contains_byte(rest, b'e') || contains_byte(rest, b'E') { - return Err(ParseFixedError::InvalidFormat); + // Single-pass accumulation + let mut acc = Accumulator::new(); + let mut i = start; + while i < bytes.len() { + acc = match acc.feed(bytes[i]) { + Ok(a) => a, + Err(e) => return Err(e), + }; + i += 1; } - // Split on decimal point - let (whole_bytes, frac_bytes) = match find_dot(rest) { - None => { - // No decimal point: integer only - if !all_digits(rest) { - return Err(ParseFixedError::InvalidFormat); - } - (rest, &[] as &[u8]) - } - Some(dot_idx) => { - let (w, after_dot) = rest.split_at(dot_idx); - // after_dot starts with '.', skip it - let f = after_dot.split_at(1).1; - - // Reject double dots - if contains_byte(f, b'.') { - return Err(ParseFixedError::InvalidFormat); - } - - // Validate digits - if !w.is_empty() && !all_digits(w) { - return Err(ParseFixedError::InvalidFormat); - } - if !f.is_empty() && !all_digits(f) { - return Err(ParseFixedError::InvalidFormat); - } - - // ".5" is valid (empty whole part), but "." alone is not - if w.is_empty() && f.is_empty() { - return Err(ParseFixedError::InvalidFormat); - } - - (w, f) - } - }; + // Must have seen at least one digit ("." alone is invalid) + if !acc.saw_any_digit { + return Err(ParseFixedError::InvalidFormat); + } - // Check excess precision - if frac_bytes.len() > D as usize { + // Check fractional precision + if acc.frac_digits > D { return Err(ParseFixedError::ExcessPrecision); } - // Parse whole part - let whole: i128 = if whole_bytes.is_empty() { - 0 + // Rescale: mantissa currently has `frac_digits` implied decimals. + // We need D decimals, so multiply by 10^(D - frac_digits). + let pad = D - acc.frac_digits; + if pad > 18 { + return Err(ParseFixedError::Overflow); + } + + // Mantissa is accumulated in negative space (always <= 0). + // For positive numbers, negate. For negative, keep as-is. + // This handles i64::MIN correctly since |-i64::MIN| > i64::MAX. + let signed_mantissa = if negative { + // Already negative — keep as-is + acc.mantissa } else { - match parse_digits(whole_bytes) { + // Negate to get positive value + match acc.mantissa.checked_neg() { Some(v) => v, None => return Err(ParseFixedError::Overflow), } }; - // Parse fractional part, zero-padding to D digits - let frac: i128 = if frac_bytes.is_empty() { - 0 - } else { - let Some(base) = parse_digits(frac_bytes) else { - return Err(ParseFixedError::Overflow); - }; - // Multiply by 10^(D - frac_len) to zero-pad - let pad = D as usize - frac_bytes.len(); - let mut result = base; - let mut p = 0; - while p < pad { - result = match result.checked_mul(10) { - Some(v) => v, - None => return Err(ParseFixedError::Overflow), - }; - p += 1; - } - result - }; - - // Combine: whole * SCALE + frac - let scale = Self::SCALE as i128; - let combined = match whole.checked_mul(scale) { - Some(v) => match v.checked_add(frac) { - Some(v2) => v2, - None => return Err(ParseFixedError::Overflow), - }, - None => return Err(ParseFixedError::Overflow), - }; - - // Apply sign - let signed = if negative { -combined } else { combined }; + if pad == 0 { + return Ok(Self::from_raw(signed_mantissa)); + } - // Narrow to i64 - if signed > (i64::MAX as i128) || signed < (i64::MIN as i128) { + let scale = POW10_I64[pad as usize]; + let Some(scaled) = signed_mantissa.checked_mul(scale) else { return Err(ParseFixedError::Overflow); - } + }; - Ok(Self::from_raw(signed as i64)) + Ok(Self::from_raw(scaled)) } /// Parse a decimal string like `"1.23"` or `"-0.5"` into `FixedI64`. /// - /// Accepts: - /// - Optional sign prefix (`-` or `+`) - /// - Integer-only: `"123"` - /// - Decimal: `"1.23"`, `".5"`, `"0.5"` - /// - Leading zeros: `"007.50"` - /// - /// Rejects: - /// - Empty strings - /// - Exponent notation (`"1e5"`) - /// - Double dots (`"1..2"`) - /// - Non-digit characters - /// - Fractional digits exceeding D + /// Delegates to [`parse_decimal_bytes`](Self::parse_decimal_bytes). /// /// # Errors /// - /// Returns [`ParseFixedError::InvalidFormat`] for malformed input, - /// [`ParseFixedError::Overflow`] if the value exceeds `i64` range, - /// or [`ParseFixedError::ExcessPrecision`] if fractional digits exceed D. + /// Same errors as [`parse_decimal_bytes`](Self::parse_decimal_bytes). pub const fn from_str_decimal(s: &str) -> Result { Self::parse_decimal_bytes(s.as_bytes()) } @@ -543,4 +503,80 @@ mod tests { assert_eq!(from_str, from_bytes, "mismatch for {input:?}"); } } + + // --- HFT price format tests --- + + #[test] + fn parse_binance_btc_price() { + // Typical Binance BTC/USDT price + let result = FixedI64::<2>::parse_decimal_bytes(b"72681.70"); + assert_eq!(result, Ok(FixedI64::<2>::from_raw(7_268_170))); + } + + #[test] + fn parse_binance_btc_qty() { + // Typical Binance BTC quantity + let result = FixedI64::<3>::parse_decimal_bytes(b"4.160"); + assert_eq!(result, Ok(FixedI64::<3>::from_raw(4160))); + } + + #[test] + fn parse_polymarket_price() { + // Typical Polymarket contract price + let result = FixedI64::<6>::parse_decimal_bytes(b"0.53"); + assert_eq!(result, Ok(FixedI64::<6>::from_raw(530_000))); + } + + #[test] + fn parse_polymarket_size() { + // Typical Polymarket order size + let result = FixedI64::<6>::parse_decimal_bytes(b"250.0"); + assert_eq!(result, Ok(FixedI64::<6>::from_raw(250_000_000))); + } + + #[test] + fn parse_max_18_digits_no_overflow() { + // 18 digits is the max we support without overflow + // 123456789012345678 fits in i64 (max ~9.2e18) + let result = FixedI64::<0>::parse_decimal_bytes(b"123456789012345678"); + assert_eq!(result, Ok(FixedI64::<0>::from_raw(123_456_789_012_345_678))); + } + + #[test] + fn parse_19_digits_fits() { + // 19 digits that fit in i64 (i64::MAX = 9223372036854775807) + let result = FixedI64::<0>::parse_decimal_bytes(b"9223372036854775807"); + assert_eq!(result, Ok(FixedI64::<0>::from_raw(i64::MAX))); + } + + #[test] + fn parse_19_digits_overflows() { + // 19 digits that exceed i64::MAX + assert_eq!( + FixedI64::<0>::parse_decimal_bytes(b"9223372036854775808"), + Err(ParseFixedError::Overflow) + ); + } + + #[test] + fn parse_20_digits_overflows() { + // 20 digits always overflows + assert_eq!( + FixedI64::<0>::parse_decimal_bytes(b"12345678901234567890"), + Err(ParseFixedError::Overflow) + ); + } + + #[test] + fn parse_many_leading_zeros() { + // 20 chars but value is 1 — leading zeros must not count toward overflow + let result = FixedI64::<0>::parse_decimal_bytes(b"00000000000000000001"); + assert_eq!(result, Ok(FixedI64::<0>::from_raw(1))); + } + + #[test] + fn parse_leading_zeros_with_decimal() { + let result = FixedI64::<6>::parse_decimal_bytes(b"000000.500000"); + assert_eq!(result, Ok(FixedI64::<6>::from_raw(500_000))); + } } diff --git a/crates/polymarket/Cargo.toml b/crates/polymarket/Cargo.toml index 7627e35..230a4e0 100644 --- a/crates/polymarket/Cargo.toml +++ b/crates/polymarket/Cargo.toml @@ -8,11 +8,12 @@ license.workspace = true repository.workspace = true [features] -default = ["simd-json"] +default = ["sonic-rs"] +sonic-rs = ["dep:sonic-rs"] simd-json = ["dep:simd-json"] [dependencies] -mantis-types = { workspace = true } +mantis-types = { workspace = true, features = ["std"] } mantis-events = { workspace = true } mantis-fixed = { workspace = true } mantis-registry = { workspace = true } @@ -21,6 +22,7 @@ mantis-transport = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } simd-json = { workspace = true, optional = true } +sonic-rs = { workspace = true, optional = true } tracing = { workspace = true } [dev-dependencies] diff --git a/crates/polymarket/README.md b/crates/polymarket/README.md new file mode 100644 index 0000000..18a3ce4 --- /dev/null +++ b/crates/polymarket/README.md @@ -0,0 +1,105 @@ +# mantis-polymarket + +Zero-allocation Polymarket CLOB WebSocket decoder for the Mantis low-latency financial SDK. + +Converts market WebSocket messages into `HotEvent` values at ~306ns per `price_change` message. + +## Architecture + +``` +COLD SIDE (transport thread) HOT SIDE (SPSC consumer) + | +WS frame -> PolymarketMarketDecoder::decode | -> HotEvent -> SPSC -> engine + peek_type() → dispatch by message type | + simd-json parse → FixedI64 | + registry.by_polymarket_token_id() | O(1) token_id → InstrumentId + meta.price_to_ticks/qty_to_lots | + emit BookDelta / Trade | + | + push() | pop_batch() +``` + +The decoder borrows an `InstrumentRegistry` for O(1) `token_id` to `InstrumentId` resolution. This supports Polymarket's market rotation model where token IDs change every window but `InstrumentId` remains stable. + +## Message Types + +| WS `type` | Output | Flags | Description | +|---|---|---|---| +| `price_change` | Single `BookDelta` | `LAST_IN_BATCH` | Level update (price + size + side) | +| `last_trade_price` | Single `Trade` | `LAST_IN_BATCH` | Executed trade with optional aggressor side | +| `book` | Batch of `BookDelta` | `IS_SNAPSHOT`, last gets `LAST_IN_BATCH` | Full book snapshot (bids + asks) | + +### Snapshot Truncation + +Book snapshots are capped at 64 levels (the output buffer size). If a snapshot exceeds 64 combined bid+ask levels, the decoder emits 64 events and logs a `tracing::warn`. This is a known limitation for deep books. + +## Usage + +```rust +use mantis_polymarket::market::*; +use mantis_registry::InstrumentRegistry; +use mantis_types::*; + +// Registry must outlive the decoder (typically 'static via Box::leak) +let registry: &'static InstrumentRegistry<6> = /* ... */; + +let mut decoder = PolymarketMarketDecoder::<6>::new( + SourceId::from_raw(10), + registry, +); + +// Decode a raw WebSocket message +let mut buf = br#"{"type":"price_change","asset_id":"abc123","price":"0.53","size":"100.0","side":"BUY"}"#.to_vec(); +let mut out = [HotEvent::heartbeat(/* ... */); 64]; +let n = decoder.decode(&mut buf, Timestamp::now(), &mut out); +// n=1, out[0] = BookDelta { price: Ticks(53), qty: Lots(100), side: Bid } +``` + +## Type Dispatch + +`peek_type()` scans raw bytes for `"type":"` and extracts the value without parsing the full JSON. This allows `simd-json`-safe dispatch: the buffer is only parsed once after the message type is known. + +## Spawning a Feed + +```rust +use mantis_polymarket::market::*; +use mantis_transport::polymarket::market::PolymarketMarketConfig; + +let result = spawn_polymarket_market_feed( + PolymarketMarketConfig { + token_ids: vec![up_token.into(), down_token.into()], + core_id: Some(1), + backoff: Default::default(), + }, + SourceId::from_raw(10), + registry, + |event| ring.try_push(event).is_ok(), +)?; + +// Monitor backpressure +let drops = result.drop_count.load(std::sync::atomic::Ordering::Relaxed); +``` + +## Key Types + +| Type | Description | +|---|---| +| `PolymarketMarketDecoder<'r, D>` | Stateful decoder borrowing an `InstrumentRegistry` | +| `FeedSpawnResult` | Handle + `event_count` + `drop_count` atomic counters | + +## Future + +- User WebSocket decoder (fills, order acks) +- CLOB REST client (order submission) +- Gamma API client (market discovery, window rotation) +- `TokenIndex` for O(1) token_id lookup without string hashing + +## Testing + +```bash +cargo +nightly test -p mantis-polymarket +``` + +## Safety + +`#![deny(unsafe_code)]` on the crate root. No unsafe blocks. diff --git a/crates/polymarket/src/market/decoder.rs b/crates/polymarket/src/market/decoder.rs index ef5594c..26ee3be 100644 --- a/crates/polymarket/src/market/decoder.rs +++ b/crates/polymarket/src/market/decoder.rs @@ -172,9 +172,20 @@ impl<'r, const D: u8> PolymarketMarketDecoder<'r, D> { } let mut count: usize = 0; + let total_levels = msg.bids.len() + msg.asks.len(); + let mut truncation_warned = false; for (depth_idx, level) in msg.bids.iter().enumerate() { if count >= 64 { + if !truncation_warned { + tracing::warn!( + asset_id = msg.asset_id, + total_levels, + emitted = 64, + "book snapshot truncated at 64 events" + ); + truncation_warned = true; + } break; } let Some((price, qty)) = parse_price_qty::(level.price, level.size, &meta) else { @@ -202,6 +213,14 @@ impl<'r, const D: u8> PolymarketMarketDecoder<'r, D> { for (depth_idx, level) in msg.asks.iter().enumerate() { if count >= 64 { + if !truncation_warned { + tracing::warn!( + asset_id = msg.asset_id, + total_levels, + emitted = 64, + "book snapshot truncated at 64 events" + ); + } break; } let Some((price, qty)) = parse_price_qty::(level.price, level.size, &meta) else { @@ -307,12 +326,17 @@ fn memchr_byte(byte: u8, haystack: &[u8]) -> Option { None } -#[cfg(feature = "simd-json")] +#[cfg(feature = "sonic-rs")] +fn parse_json<'a, T: serde::Deserialize<'a>>(buf: &'a mut [u8]) -> Result { + sonic_rs::from_slice(&*buf).map_err(|_| ()) +} + +#[cfg(all(feature = "simd-json", not(feature = "sonic-rs")))] fn parse_json<'a, T: serde::Deserialize<'a>>(buf: &'a mut [u8]) -> Result { simd_json::from_slice(buf).map_err(|_| ()) } -#[cfg(not(feature = "simd-json"))] +#[cfg(not(any(feature = "sonic-rs", feature = "simd-json")))] fn parse_json<'a, T: serde::Deserialize<'a>>(buf: &'a mut [u8]) -> Result { serde_json::from_slice(buf).map_err(|_| ()) } @@ -711,6 +735,48 @@ mod tests { } } + #[test] + fn decode_book_truncates_at_64_levels() { + use std::fmt::Write as _; + + let (reg, _) = test_registry(); + let mut decoder = PolymarketMarketDecoder::<6>::new(SourceId::from_raw(10), ®); + let mut out = make_out(); + + // Build a book with 70 levels (40 bids + 30 asks) to trigger truncation + let mut json = String::from(r#"{"type":"book","asset_id":"abc123","bids":["#); + for i in 0..40_u32 { + if i > 0 { + json.push(','); + } + let _ = write!( + json, + r#"{{"price":"0.{:02}","size":"{}"}}"#, + 50_u32.saturating_sub(i), + 100 + i + ); + } + json.push_str(r#"],"asks":["#); + for i in 0..30_u32 { + if i > 0 { + json.push(','); + } + let _ = write!( + json, + r#"{{"price":"0.{:02}","size":"{}"}}"#, + 55 + i, + 100 + i + ); + } + json.push_str("]}"); + + let mut buf = json.into_bytes(); + let recv_ts = Timestamp::from_nanos(1000); + let n = decoder.decode(&mut buf, recv_ts, &mut out); + + assert_eq!(n, 64, "should truncate at 64 events"); + } + #[test] fn peek_type_truncated_returns_none() { // JSON with "type":" present but no closing quote for the value diff --git a/crates/polymarket/src/market/mod.rs b/crates/polymarket/src/market/mod.rs index f70a41e..2b74e80 100644 --- a/crates/polymarket/src/market/mod.rs +++ b/crates/polymarket/src/market/mod.rs @@ -5,3 +5,4 @@ pub(crate) mod schema; pub mod spawn; pub use decoder::PolymarketMarketDecoder; +pub use spawn::{FeedSpawnResult, spawn_polymarket_market_feed}; diff --git a/crates/polymarket/src/market/spawn.rs b/crates/polymarket/src/market/spawn.rs index 729b1b4..e8996bf 100644 --- a/crates/polymarket/src/market/spawn.rs +++ b/crates/polymarket/src/market/spawn.rs @@ -1,3 +1,175 @@ //! Convenience functions for spawning Polymarket market feed threads. -//! -//! Will be implemented in a later task. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use mantis_events::HotEvent; +use mantis_registry::InstrumentRegistry; +use mantis_transport::FeedHandle; +use mantis_transport::polymarket::market::PolymarketMarketConfig; +use mantis_types::{SourceId, Timestamp}; + +use crate::market::decoder::PolymarketMarketDecoder; + +/// Result of spawning a Polymarket market feed. +pub struct FeedSpawnResult { + /// Handle for shutdown and monitoring. + pub handle: FeedHandle, + /// Counter of received messages / callback invocations (for `FeedMonitor` liveness). + pub event_count: Arc, + /// Counter of events dropped due to push returning false. + pub drop_count: Arc, +} + +/// Build the transport callback (exposed for testing). +pub fn build_callback( + mut decoder: PolymarketMarketDecoder<'static, D>, + mut push: F, +) -> ( + impl FnMut(&mut [u8]) -> bool, + Arc, + Arc, +) +where + F: FnMut(HotEvent) -> bool + Send + 'static, +{ + let event_count = Arc::new(AtomicU64::new(0)); + let drop_count = Arc::new(AtomicU64::new(0)); + let ec = Arc::clone(&event_count); + let dc = Arc::clone(&drop_count); + + let mut out = make_out(); + let callback = move |buf: &mut [u8]| { + // Always count — proves the feed is alive even for non-hot messages + ec.fetch_add(1, Ordering::Relaxed); + let recv_ts = Timestamp::now(); + let count = decoder.decode(buf, recv_ts, &mut out); + for event in out.iter().take(count) { + if !push(*event) { + dc.fetch_add(1, Ordering::Relaxed); + } + } + true + }; + + (callback, event_count, drop_count) +} + +/// Spawn a Polymarket market feed thread with built-in decoder. +/// +/// # Errors +/// +/// Returns an error if the OS fails to spawn the feed thread or if no +/// token IDs are configured. +pub fn spawn_polymarket_market_feed( + config: PolymarketMarketConfig, + source_id: SourceId, + registry: &'static InstrumentRegistry, + push: F, +) -> Result +where + F: FnMut(HotEvent) -> bool + Send + 'static, +{ + let decoder = PolymarketMarketDecoder::new(source_id, registry); + let (callback, event_count, drop_count) = build_callback(decoder, push); + let handle = mantis_transport::polymarket::market::spawn_market_feed(config, callback)?; + Ok(FeedSpawnResult { + handle, + event_count, + drop_count, + }) +} + +fn make_out() -> [HotEvent; 64] { + [HotEvent::heartbeat( + Timestamp::ZERO, + mantis_types::SeqNum::ZERO, + mantis_types::SourceId::from_raw(0), + mantis_events::EventFlags::EMPTY, + mantis_events::HeartbeatPayload { counter: 0 }, + ); 64] +} + +#[cfg(test)] +mod tests { + use super::*; + use mantis_fixed::FixedI64; + use mantis_registry::*; + use mantis_types::{InstrumentMeta, Timestamp}; + + /// Create a leaked static registry with one BTC-M15-Up instrument. + /// Follows the exact pattern from polymarket decoder tests. + #[expect(clippy::expect_used, reason = "test-only helper")] + fn test_registry() -> &'static InstrumentRegistry<6> { + let mut reg = InstrumentRegistry::<6>::new(); + let meta = InstrumentMeta::new( + FixedI64::<6>::from_raw(10_000), // tick_size = 0.01 + FixedI64::<6>::from_raw(1_000_000), // lot_size = 1.0 + ) + .expect("valid meta"); + + let key = InstrumentKey::prediction(Asset::Btc, Timeframe::M15, OutcomeSide::Up); + let id = reg + .insert(key, meta, None, Some(PolymarketBinding::default())) + .expect("insert ok"); + reg.bind_polymarket_current( + id, + PolymarketWindowBinding { + token_id: "abc123".to_owned(), + market_slug: "btc-15m-up".to_owned(), + window_start: Timestamp::from_nanos(0), + window_end: Timestamp::from_nanos(900_000_000_000), + condition_id: None, + }, + ) + .expect("bind ok"); + Box::leak(Box::new(reg)) + } + + #[test] + fn callback_decodes_and_pushes() { + let registry = test_registry(); + let decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + + let push_count = Arc::new(AtomicU64::new(0)); + let pc = Arc::clone(&push_count); + + let (mut callback, event_count, drop_count) = build_callback(decoder, move |_| { + pc.fetch_add(1, Ordering::Relaxed); + true + }); + + let mut buf = br#"{"type":"price_change","asset_id":"abc123","price":"0.53","size":"100.0","side":"BUY"}"#.to_vec(); + assert!(callback(&mut buf)); + assert_eq!(push_count.load(Ordering::Relaxed), 1); + assert_eq!(event_count.load(Ordering::Relaxed), 1); + assert_eq!(drop_count.load(Ordering::Relaxed), 0); + } + + #[test] + fn make_out_returns_64_events() { + let out = make_out(); + assert_eq!(out.len(), 64); + } + + #[test] + fn feed_spawn_result_fields_accessible() { + let handle_event_count = Arc::new(AtomicU64::new(42)); + let handle_drop_count = Arc::new(AtomicU64::new(7)); + assert_eq!(handle_event_count.load(Ordering::Relaxed), 42); + assert_eq!(handle_drop_count.load(Ordering::Relaxed), 7); + } + + #[test] + fn callback_counts_drops() { + let registry = test_registry(); + let decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + + let (mut callback, event_count, drop_count) = build_callback(decoder, |_| false); + + let mut buf = br#"{"type":"price_change","asset_id":"abc123","price":"0.53","size":"100.0","side":"BUY"}"#.to_vec(); + callback(&mut buf); + assert_eq!(event_count.load(Ordering::Relaxed), 1); + assert_eq!(drop_count.load(Ordering::Relaxed), 1); + } +} diff --git a/crates/polymarket/tests/e2e_decode.rs b/crates/polymarket/tests/e2e_decode.rs new file mode 100644 index 0000000..fd019e2 --- /dev/null +++ b/crates/polymarket/tests/e2e_decode.rs @@ -0,0 +1,457 @@ +//! End-to-end and stress tests for the Polymarket decode pipeline. +//! +//! Tests the full path: raw JSON bytes → `PolymarketMarketDecoder` → `HotEvent` output. +//! Includes stress tests for sustained message rates and batch book snapshots. + +#![expect(clippy::print_stderr, reason = "stress-test diagnostics use eprintln")] +#![expect( + clippy::cast_precision_loss, + reason = "f64 stats in test diagnostics are fine" +)] + +use mantis_events::{EventBody, EventFlags, HeartbeatPayload, HotEvent}; +use mantis_fixed::FixedI64; +use mantis_polymarket::market::PolymarketMarketDecoder; +use mantis_registry::*; +use mantis_types::{InstrumentId, InstrumentMeta, Lots, SeqNum, Side, SourceId, Ticks, Timestamp}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Instant; + +fn make_out() -> [HotEvent; 64] { + [HotEvent::heartbeat( + Timestamp::ZERO, + SeqNum::ZERO, + SourceId::from_raw(0), + EventFlags::EMPTY, + HeartbeatPayload { counter: 0 }, + ); 64] +} + +#[expect( + clippy::expect_used, + reason = "test-only helper with known-valid constants" +)] +fn test_registry() -> &'static InstrumentRegistry<6> { + let mut reg = InstrumentRegistry::<6>::new(); + let meta = InstrumentMeta::new( + FixedI64::<6>::from_raw(10_000), // tick_size = 0.01 + FixedI64::<6>::from_raw(1_000_000), // lot_size = 1.0 + ) + .expect("valid meta"); + + // BTC-M15-Up + let key_up = InstrumentKey::prediction(Asset::Btc, Timeframe::M15, OutcomeSide::Up); + let id_up = reg + .insert(key_up, meta, None, Some(PolymarketBinding::default())) + .expect("insert up"); + reg.bind_polymarket_current( + id_up, + PolymarketWindowBinding { + token_id: "token_up_123".to_owned(), + market_slug: "btc-15m-up".to_owned(), + window_start: Timestamp::from_nanos(0), + window_end: Timestamp::from_nanos(900_000_000_000), + condition_id: None, + }, + ) + .expect("bind up"); + + // BTC-M15-Down + let key_down = InstrumentKey::prediction(Asset::Btc, Timeframe::M15, OutcomeSide::Down); + let id_down = reg + .insert(key_down, meta, None, Some(PolymarketBinding::default())) + .expect("insert down"); + reg.bind_polymarket_current( + id_down, + PolymarketWindowBinding { + token_id: "token_down_456".to_owned(), + market_slug: "btc-15m-down".to_owned(), + window_start: Timestamp::from_nanos(0), + window_end: Timestamp::from_nanos(900_000_000_000), + condition_id: None, + }, + ) + .expect("bind down"); + + Box::leak(Box::new(reg)) +} + +// --- End-to-End Tests --- + +/// Full pipeline: `price_change` → `BookDelta` with all fields verified. +#[test] +#[expect(clippy::panic, reason = "test uses panic for match-arm failure")] +fn e2e_price_change_full_verification() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + let mut buf = + br#"{"type":"price_change","asset_id":"token_up_123","price":"0.53","size":"250.0","side":"BUY"}"# + .to_vec(); + let recv_ts = Timestamp::from_nanos(1_000_000_000); + let n = decoder.decode(&mut buf, recv_ts, &mut out); + + assert_eq!(n, 1); + let event = &out[0]; + assert_eq!(event.header.recv_ts, recv_ts); + assert_eq!(event.header.source_id, SourceId::from_raw(1)); + assert_eq!(event.header.instrument_id, InstrumentId::from_raw(1)); + + match &event.body { + EventBody::BookDelta(bd) => { + assert_eq!(bd.price, Ticks::from_raw(53)); // 0.53 / 0.01 + assert_eq!(bd.qty, Lots::from_raw(250)); // 250.0 / 1.0 + assert_eq!(bd.side, Side::Bid); + } + other => panic!("expected BookDelta, got {:?}", other.kind()), + } +} + +/// Multi-token routing: Up and Down tokens map to different instruments. +#[test] +fn e2e_multi_token_routing() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + // Up token + let mut buf_up = + br#"{"type":"price_change","asset_id":"token_up_123","price":"0.60","size":"100.0","side":"BUY"}"# + .to_vec(); + let n = decoder.decode(&mut buf_up, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 1); + assert_eq!(out[0].header.instrument_id, InstrumentId::from_raw(1)); + + // Down token + let mut buf_down = + br#"{"type":"price_change","asset_id":"token_down_456","price":"0.40","size":"200.0","side":"SELL"}"# + .to_vec(); + let n = decoder.decode(&mut buf_down, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 1); + assert_eq!(out[0].header.instrument_id, InstrumentId::from_raw(2)); +} + +/// Book snapshot produces batch with `LAST_IN_BATCH` on final event. +#[test] +#[expect(clippy::panic, reason = "test uses panic for match-arm failure")] +fn e2e_book_snapshot_batch() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + let mut buf = br#"{"type":"book","asset_id":"token_up_123","bids":[{"price":"0.50","size":"100"},{"price":"0.49","size":"200"}],"asks":[{"price":"0.55","size":"150"},{"price":"0.56","size":"80"}]}"#.to_vec(); + + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 4); + + // All should have IS_SNAPSHOT + for (i, ev) in out[..4].iter().enumerate() { + assert!( + ev.header.flags.contains(EventFlags::IS_SNAPSHOT), + "event {i} missing IS_SNAPSHOT" + ); + } + + // Only last should have LAST_IN_BATCH + for (i, ev) in out[..3].iter().enumerate() { + assert!( + !ev.header.flags.contains(EventFlags::LAST_IN_BATCH), + "event {i} should NOT have LAST_IN_BATCH" + ); + } + assert!( + out[3].header.flags.contains(EventFlags::LAST_IN_BATCH), + "last event should have LAST_IN_BATCH" + ); + + // Verify sides + assert_eq!( + match &out[0].body { + EventBody::BookDelta(bd) => bd.side, + _ => panic!("expected BookDelta"), + }, + Side::Bid + ); + assert_eq!( + match &out[2].body { + EventBody::BookDelta(bd) => bd.side, + _ => panic!("expected BookDelta"), + }, + Side::Ask + ); +} + +/// Trade event from `last_trade_price`. +#[test] +#[expect(clippy::panic, reason = "test uses panic for match-arm failure")] +fn e2e_trade_event() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + let mut buf = + br#"{"type":"last_trade_price","asset_id":"token_up_123","price":"0.55","size":"50.0","side":"SELL"}"# + .to_vec(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 1); + + match &out[0].body { + EventBody::Trade(t) => { + assert_eq!(t.price, Ticks::from_raw(55)); + assert_eq!(t.qty, Lots::from_raw(50)); + assert_eq!(t.aggressor, Side::Ask); + } + other => panic!("expected Trade, got {:?}", other.kind()), + } +} + +/// Non-hot message types are skipped. +#[test] +fn e2e_non_hot_types_skipped() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + for msg_type in &[ + "best_bid_ask", + "tick_size_change", + "market_resolved", + "new_market", + ] { + let mut buf = format!(r#"{{"type":"{msg_type}"}}"#).into_bytes(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 0, "{msg_type} should be skipped"); + } +} + +/// Unknown token returns 0 events. +#[test] +fn e2e_unknown_token() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + let mut buf = + br#"{"type":"price_change","asset_id":"unknown_token_xyz","price":"0.50","size":"100","side":"BUY"}"# + .to_vec(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 0); +} + +/// Malformed inputs never panic. +#[test] +fn e2e_malformed_never_panics() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + let bad: Vec> = vec![ + b"".to_vec(), + b"not json".to_vec(), + b"{}".to_vec(), + b"{\"type\":\"price_change\"}".to_vec(), + b"\x00\xff\xfe".to_vec(), + vec![b'{'; 65536], + ]; + + for (i, input) in bad.iter().enumerate() { + let mut buf = input.clone(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 0, "bad input {i} should produce 0 events"); + } +} + +// --- Stress Tests --- + +/// 1M `price_change` messages sustained throughput. +#[test] +#[ignore = "stress test"] +fn stress_1m_price_changes() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + let template = + br#"{"type":"price_change","asset_id":"token_up_123","price":"0.53","size":"100.0","side":"BUY"}"#; + + let iterations = 1_000_000u64; + let mut total_events = 0u64; + + let start = Instant::now(); + for _ in 0..iterations { + let mut buf = template.to_vec(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + total_events += n as u64; + } + let elapsed = start.elapsed(); + + assert_eq!(total_events, iterations); + + let ns_per_msg = elapsed.as_nanos() as f64 / iterations as f64; + let msgs_per_sec = iterations as f64 / elapsed.as_secs_f64(); + + eprintln!("--- Polymarket Price Change Stress Test ---"); + eprintln!(" Messages: {iterations}"); + eprintln!(" Total time: {elapsed:?}"); + eprintln!(" ns/msg: {ns_per_msg:.1}"); + eprintln!(" msgs/sec: {msgs_per_sec:.0}"); + + assert!( + ns_per_msg < 10_000.0, + "decode too slow: {ns_per_msg:.0}ns/msg" + ); +} + +/// Alternating Up/Down tokens for multi-instrument stress. +#[test] +#[ignore = "stress test"] +fn stress_500k_alternating_tokens() { + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + let up_msg = + br#"{"type":"price_change","asset_id":"token_up_123","price":"0.60","size":"100","side":"BUY"}"#; + let down_msg = + br#"{"type":"price_change","asset_id":"token_down_456","price":"0.40","size":"100","side":"SELL"}"#; + + let iterations = 500_000u64; + let mut up_count = 0u64; + let mut down_count = 0u64; + + let start = Instant::now(); + for i in 0..iterations { + let mut buf = if i % 2 == 0 { + up_msg.to_vec() + } else { + down_msg.to_vec() + }; + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 1); + if out[0].header.instrument_id == InstrumentId::from_raw(1) { + up_count += 1; + } else { + down_count += 1; + } + } + let elapsed = start.elapsed(); + + assert_eq!(up_count, iterations / 2); + assert_eq!(down_count, iterations / 2); + + let ns_per_msg = elapsed.as_nanos() as f64 / iterations as f64; + eprintln!("--- Polymarket Multi-Token Stress Test ---"); + eprintln!(" Messages: {iterations} ({up_count} Up + {down_count} Down)"); + eprintln!(" Total time: {elapsed:?}"); + eprintln!(" ns/msg: {ns_per_msg:.1}"); +} + +/// Book snapshot stress: 10K snapshots with 20 levels each. +#[test] +#[ignore = "stress test"] +#[expect(clippy::expect_used, reason = "write!() to String is infallible")] +fn stress_10k_book_snapshots() { + use std::fmt::Write; + + let registry = test_registry(); + let mut decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + let mut out = make_out(); + + // Build a 20-level snapshot + let mut json = String::from(r#"{"type":"book","asset_id":"token_up_123","bids":["#); + for i in 0..10 { + if i > 0 { + json.push(','); + } + write!( + json, + r#"{{"price":"0.{:02}","size":"{}"}}"#, + 50 - i, + 100 + i * 10 + ) + .expect("write to String"); + } + json.push_str(r#"],"asks":["#); + for i in 0..10 { + if i > 0 { + json.push(','); + } + write!( + json, + r#"{{"price":"0.{:02}","size":"{}"}}"#, + 55 + i, + 100 + i * 10 + ) + .expect("write to String"); + } + json.push_str("]}"); + let template = json.into_bytes(); + + let iterations = 10_000u64; + let mut total_events = 0u64; + + let start = Instant::now(); + for _ in 0..iterations { + let mut buf = template.clone(); + let n = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + assert_eq!(n, 20); + total_events += n as u64; + } + let elapsed = start.elapsed(); + + assert_eq!(total_events, iterations * 20); + + let ns_per_snapshot = elapsed.as_nanos() as f64 / iterations as f64; + let ns_per_event = elapsed.as_nanos() as f64 / total_events as f64; + eprintln!("--- Polymarket Book Snapshot Stress Test ---"); + eprintln!(" Snapshots: {iterations} (20 levels each)"); + eprintln!(" Total events: {total_events}"); + eprintln!(" Total time: {elapsed:?}"); + eprintln!(" ns/snapshot: {ns_per_snapshot:.1}"); + eprintln!(" ns/event: {ns_per_event:.1}"); +} + +/// Spawn callback with backpressure simulation. +#[test] +#[ignore = "stress test"] +fn stress_spawn_callback_backpressure() { + use mantis_polymarket::market::spawn::build_callback; + + let registry = test_registry(); + let decoder = PolymarketMarketDecoder::new(SourceId::from_raw(1), registry); + + let accept_count = Arc::new(AtomicU64::new(0)); + let ac = Arc::clone(&accept_count); + + let (mut callback, event_count, drop_count) = build_callback(decoder, move |_| { + let n = ac.fetch_add(1, Ordering::Relaxed); + !n.is_multiple_of(3) // drop every 3rd event (~33% drop rate) + }); + + let template = + br#"{"type":"price_change","asset_id":"token_up_123","price":"0.53","size":"100","side":"BUY"}"#; + + let iterations = 100_000u64; + for _ in 0..iterations { + let mut buf = template.to_vec(); + callback(&mut buf); + } + + let events = event_count.load(Ordering::Relaxed); + let drops = drop_count.load(Ordering::Relaxed); + + eprintln!("--- Polymarket Backpressure Stress Test ---"); + eprintln!(" Iterations: {iterations}"); + eprintln!(" Events: {events}"); + eprintln!(" Drops: {drops}"); + eprintln!(" Drop rate: {:.1}%", drops as f64 / events as f64 * 100.0); + + assert_eq!(events, iterations); + assert!(drops > 0); + assert!( + drops > iterations / 5, + "expected ~33% drops, got {drops}/{events}" + ); +} diff --git a/crates/transport/README.md b/crates/transport/README.md index 0f46940..cea50c2 100644 --- a/crates/transport/README.md +++ b/crates/transport/README.md @@ -1,6 +1,6 @@ # mantis-transport -WebSocket transport ingest layer for the Mantis low-latency financial SDK. +WebSocket transport, timer, and feed monitoring infrastructure for the Mantis low-latency financial SDK. Blocking IO on dedicated pinned threads. No async runtime. No Mutex. @@ -9,26 +9,30 @@ Blocking IO on dedicated pinned threads. No async runtime. No Mutex. ``` VENUE FEEDS MANTIS HOT PATH -Polymarket Market WS ──> [SPSC 4096] ──┐ -Polymarket User WS ──> [SPSC 1024] ──┤──> market-state engine -Binance Reference WS ──> [SPSC 8192] ──┤ -Timer ──> [SPSC 256] ──┘ +Polymarket Market WS ──> callback ──┐ +Polymarket User WS ──> callback ──┤──> venue decoder ──> [SPSC] ──> engine +Binance Reference WS ──> callback ──┤ +Timer thread ──> [SPSC 256] ──────────────────────────────┘ + │ +FeedMonitor ◄── event_count atomics ────┘ (stale feed detection) ``` Each feed runs on a dedicated CPU-pinned thread. The thread calls -`tungstenite::WebSocket::read()` in a blocking loop, parses venue JSON -into `HotEvent` values, and pushes them into `SpscRingCopy` -queues. +`tungstenite::WebSocket::read()` in a blocking loop and delivers raw +`&mut [u8]` buffers to an `FnMut` callback. Venue-specific JSON parsing +and `HotEvent` emission live in separate crates (`mantis-binance`, +`mantis-polymarket`), not in transport. ## Thread Model | Layer | What happens | Allowed | |---|---|---| -| IO threads (this crate) | WS + TLS, JSON parse, `FixedI64` → `Ticks`/`Lots`, emit `HotEvent` | alloc, strings, serde | +| IO threads (this crate) | WS + TLS, deliver raw bytes via callback | alloc, strings, network | +| Venue decoders (separate crates) | JSON parse, `FixedI64` -> `Ticks`/`Lots`, emit `HotEvent` | alloc, serde | | Owner threads (engine) | book update, signal, order decision | no alloc, no strings, no JSON | The SPSC ring is the boundary. After `push()`, no strings or decimal types -exist — only `InstrumentId`, `Ticks`, `Lots`, and compact enums. +exist -- only `InstrumentId`, `Ticks`, `Lots`, and compact enums. ## Venue Adapters @@ -44,7 +48,7 @@ let handle = spawn_market_feed( backoff: BackoffConfig::default(), }, |msg| { - // raw JSON: book, price_change, last_trade_price, etc. + // raw &mut [u8]: book, price_change, last_trade_price, etc. true // continue }, )?; @@ -62,7 +66,7 @@ use mantis_transport::binance::reference::*; let handle = spawn_reference_feed( BinanceReferenceConfig::default(), // btcusdt@bookTicker |msg| { - // raw JSON: {"e":"bookTicker","b":"67396.70","B":"8.819",...} + // raw &mut [u8]: {"e":"bookTicker","b":"67396.70","B":"8.819",...} true }, )?; @@ -72,20 +76,58 @@ let handle = spawn_reference_feed( - No subscription message needed (stream selection in URL path) - Uses futures endpoint (`fstream`) for broader geo-availability -## Feed Lifecycle +## Timer Thread + +The timer thread emits periodic `HotEvent::Timer` and `HotEvent::Heartbeat` events at configurable intervals. It runs on a dedicated thread with no WS connection. +```rust +use mantis_transport::{TimerConfig, TimerThread}; + +let timer = TimerThread::spawn( + TimerConfig { + name: "timer".into(), + tick_interval: Duration::from_millis(100), + heartbeat_interval: Duration::from_secs(1), + source_id: SourceId::from_raw(99), + core_id: Some(3), + }, + |event| { let _ = ring.try_push(event); }, +)?; ``` -spawn() → connect → subscribe → read loop ──> callback - │ - on error: reconnect with backoff (1s → 30s, ±12.5% jitter) - │ - shutdown() → responds within 100ms, thread joins + +## Feed Monitor + +`FeedMonitor` passively tracks `event_count` atomics from feed spawn wrappers and detects feeds that have stopped producing events. The engine calls `check_all()` on each `Timer(Periodic)` event. + +```rust +use mantis_transport::FeedMonitor; + +let mut monitor = FeedMonitor::new(); +monitor.register(source_id, event_count_arc)?; + +// On each timer tick: +let stale_count = monitor.check_all(); +if stale_count > 0 { + for info in monitor.stale_feeds() { + // info.source_id, info.last_event_count + } +} +``` + +## Feed Lifecycle + +```text +spawn() -> connect -> subscribe -> read loop --> callback + | + on error: reconnect with backoff (1s -> 30s, +/-12.5% jitter) + | + shutdown() -> responds within 100ms, thread joins ``` Monitoring counters (lock-free `AtomicU64`): -- `msg_count` — messages delivered to callback -- `reconnects` — successful reconnections -- `drops` — messages dropped (future: backpressure) +- `msg_count` -- messages delivered to callback +- `reconnects` -- successful reconnections +- `drops` -- messages dropped (future: backpressure) ## Socket Tuning @@ -97,7 +139,6 @@ Monitoring counters (lock-free `AtomicU64`): | Feature | Effect | |---|---| -| `simd-json` (default) | SIMD-accelerated JSON parsing for Phase C normalization | | `tuning` (default) | Linux socket tuning (`SO_BUSY_POLL`) | | `live-tests` | Integration tests against real exchange endpoints | diff --git a/crates/transport/src/feed.rs b/crates/transport/src/feed.rs index be18092..22c0115 100644 --- a/crates/transport/src/feed.rs +++ b/crates/transport/src/feed.rs @@ -165,7 +165,7 @@ impl FeedThread { /// Sleep for `duration`, checking the shutdown flag every 100ms. /// Returns `true` if shutdown was requested during the sleep. -fn interruptible_sleep(duration: Duration, shutdown: &AtomicBool) -> bool { +pub(crate) fn interruptible_sleep(duration: Duration, shutdown: &AtomicBool) -> bool { let deadline = std::time::Instant::now() + duration; let chunk = Duration::from_millis(100); while std::time::Instant::now() < deadline { diff --git a/crates/transport/src/lib.rs b/crates/transport/src/lib.rs index cccaf1c..6c5875c 100644 --- a/crates/transport/src/lib.rs +++ b/crates/transport/src/lib.rs @@ -1,9 +1,8 @@ -//! WebSocket transport ingest layer for the Mantis SDK. +//! WebSocket transport and timing infrastructure for the Mantis SDK. //! -//! This crate connects to venue WebSocket feeds (Polymarket, Binance), -//! parses JSON into [`mantis_events::HotEvent`] values, and pushes them -//! into [`mantis_queue::SpscRingCopy`] queues for consumption by -//! downstream engine threads. +//! This crate provides raw WebSocket feed threads (Polymarket, Binance) +//! and a timer thread for periodic events. Venue-specific JSON decoding +//! lives in `mantis-binance` and `mantis-polymarket` crates. //! //! # Architecture //! @@ -22,10 +21,14 @@ pub mod binance; mod feed; +mod monitor; pub mod polymarket; +mod timer; mod tuning; mod ws; pub use feed::{BackoffConfig, FeedConfig, FeedHandle, FeedThread}; +pub use monitor::{FeedMonitor, MAX_FEEDS, MonitorFullError, StaleFeedInfo}; +pub use timer::{TimerConfig, TimerConfigError, TimerSpawnError, TimerThread}; pub use tuning::SocketTuning; pub use ws::{WsConfig, WsConnection, WsError}; diff --git a/crates/transport/src/monitor.rs b/crates/transport/src/monitor.rs new file mode 100644 index 0000000..796a82a --- /dev/null +++ b/crates/transport/src/monitor.rs @@ -0,0 +1,281 @@ +//! Feed liveness monitor. +//! +//! Tracks event emission counters from feed spawn wrappers and detects +//! staleness. The engine calls [`FeedMonitor::check_all`] on each +//! `Timer(Periodic)` event. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use mantis_types::SourceId; + +/// Maximum number of feeds that can be monitored. +pub const MAX_FEEDS: usize = 8; + +/// Feed liveness monitor. +/// +/// Passively tracks event emission counters and detects feeds that have +/// stopped producing events since the last check cycle. +pub struct FeedMonitor { + feeds: [Option; MAX_FEEDS], + len: usize, +} + +struct FeedState { + event_count: Arc, + last_seen: u64, + source_id: SourceId, + stale: bool, + first_check: bool, +} + +/// Information about a stale feed. +#[derive(Debug, Clone, Copy)] +pub struct StaleFeedInfo { + /// Source identifier of the stale feed. + pub source_id: SourceId, + /// Last observed event count. + pub last_event_count: u64, +} + +/// Error when registering more feeds than [`MAX_FEEDS`]. +#[derive(Debug)] +pub struct MonitorFullError; + +impl std::fmt::Display for MonitorFullError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("feed monitor is full") + } +} + +impl std::error::Error for MonitorFullError {} + +impl FeedMonitor { + /// Creates a new empty feed monitor. + #[must_use] + pub const fn new() -> Self { + Self { + feeds: [None, None, None, None, None, None, None, None], + len: 0, + } + } + + /// Registers a feed for monitoring. + /// + /// The `event_count` should be the shared atomic counter from the feed + /// spawn result. The first [`check_all`](Self::check_all) call after + /// registration establishes a baseline and will not mark the feed stale. + /// + /// # Errors + /// + /// Returns [`MonitorFullError`] if [`MAX_FEEDS`] feeds are already registered. + pub fn register( + &mut self, + source_id: SourceId, + event_count: Arc, + ) -> Result<(), MonitorFullError> { + if self.len >= MAX_FEEDS { + return Err(MonitorFullError); + } + let current = event_count.load(Ordering::Relaxed); + self.feeds[self.len] = Some(FeedState { + event_count, + last_seen: current, + source_id, + stale: false, + first_check: true, + }); + self.len += 1; + Ok(()) + } + + /// Checks all registered feeds for staleness. + /// + /// For each feed, compares the current event count against the last + /// observed value. On the first check after registration, this + /// establishes a baseline without marking the feed stale. + /// + /// Returns the number of currently stale feeds. + pub fn check_all(&mut self) -> usize { + let mut stale_count = 0; + for slot in &mut self.feeds[..self.len] { + let Some(feed) = slot.as_mut() else { + continue; + }; + let current = feed.event_count.load(Ordering::Relaxed); + if feed.first_check { + feed.first_check = false; + feed.last_seen = current; + continue; + } + if current == feed.last_seen { + feed.stale = true; + stale_count += 1; + } else { + feed.stale = false; + } + feed.last_seen = current; + } + stale_count + } + + /// Returns an iterator over currently stale feeds. + pub fn stale_feeds(&self) -> impl Iterator + '_ { + self.feeds[..self.len].iter().filter_map(|slot| { + let feed = slot.as_ref()?; + if feed.stale { + Some(StaleFeedInfo { + source_id: feed.source_id, + last_event_count: feed.last_seen, + }) + } else { + None + } + }) + } + + /// Checks whether a specific feed is currently marked stale. + #[must_use] + pub fn is_stale(&self, source_id: SourceId) -> bool { + self.feeds[..self.len].iter().any(|slot| { + slot.as_ref() + .is_some_and(|f| f.source_id == source_id && f.stale) + }) + } + + /// Returns the number of registered feeds. + #[must_use] + pub fn len(&self) -> usize { + self.len + } + + /// Returns `true` if no feeds are registered. + #[must_use] + pub fn is_empty(&self) -> bool { + self.len == 0 + } +} + +impl Default for FeedMonitor { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn counter(val: u64) -> Arc { + Arc::new(AtomicU64::new(val)) + } + + #[test] + fn new_monitor_is_empty() { + let mon = FeedMonitor::new(); + assert_eq!(mon.len(), 0); + assert!(mon.is_empty()); + } + + #[test] + fn first_check_establishes_baseline() { + let cnt = counter(0); + let mut mon = FeedMonitor::new(); + mon.register(SourceId::from_raw(1), Arc::clone(&cnt)).ok(); + let stale = mon.check_all(); + assert_eq!(stale, 0); + assert!(!mon.is_stale(SourceId::from_raw(1))); + } + + #[test] + fn unchanged_after_baseline_is_stale() { + let cnt = counter(0); + let mut mon = FeedMonitor::new(); + mon.register(SourceId::from_raw(1), Arc::clone(&cnt)).ok(); + mon.check_all(); // baseline + let stale = mon.check_all(); // no change + assert_eq!(stale, 1); + assert!(mon.is_stale(SourceId::from_raw(1))); + } + + #[test] + fn active_feed_not_stale() { + let cnt = counter(0); + let mut mon = FeedMonitor::new(); + mon.register(SourceId::from_raw(1), Arc::clone(&cnt)).ok(); + mon.check_all(); // baseline + cnt.fetch_add(5, Ordering::Relaxed); + let stale = mon.check_all(); + assert_eq!(stale, 0); + assert!(!mon.is_stale(SourceId::from_raw(1))); + } + + #[test] + fn recovery_clears_stale() { + let cnt = counter(0); + let mut mon = FeedMonitor::new(); + mon.register(SourceId::from_raw(1), Arc::clone(&cnt)).ok(); + mon.check_all(); // baseline + assert_eq!(mon.check_all(), 1); // stale + assert!(mon.is_stale(SourceId::from_raw(1))); + cnt.fetch_add(1, Ordering::Relaxed); + assert_eq!(mon.check_all(), 0); // recovered + assert!(!mon.is_stale(SourceId::from_raw(1))); + } + + #[test] + fn multiple_feeds() { + let cnt_a = counter(0); + let cnt_b = counter(0); + let mut mon = FeedMonitor::new(); + mon.register(SourceId::from_raw(1), Arc::clone(&cnt_a)).ok(); + mon.register(SourceId::from_raw(2), Arc::clone(&cnt_b)).ok(); + mon.check_all(); // baseline for both + cnt_a.fetch_add(1, Ordering::Relaxed); + // cnt_b unchanged + let stale = mon.check_all(); + assert_eq!(stale, 1); + assert!(!mon.is_stale(SourceId::from_raw(1))); + assert!(mon.is_stale(SourceId::from_raw(2))); + } + + #[test] + fn max_feeds_exceeded() { + let mut mon = FeedMonitor::new(); + for i in 0..MAX_FEEDS { + #[expect(clippy::cast_possible_truncation)] + let id = SourceId::from_raw(i as u16); + assert!(mon.register(id, counter(0)).is_ok()); + } + #[expect(clippy::cast_possible_truncation)] + let overflow_id = SourceId::from_raw(MAX_FEEDS as u16); + assert!(mon.register(overflow_id, counter(0)).is_err()); + } + + #[test] + fn monitor_full_error_display() { + let e = MonitorFullError; + assert!(!e.to_string().is_empty()); + assert_eq!(e.to_string(), "feed monitor is full"); + } + + #[test] + fn monitor_full_error_is_std_error() { + let e: &dyn std::error::Error = &MonitorFullError; + assert!(e.source().is_none()); + } + + #[test] + fn stale_feeds_iterator() { + let cnt_a = counter(0); + let cnt_b = counter(0); + let mut mon = FeedMonitor::new(); + mon.register(SourceId::from_raw(1), Arc::clone(&cnt_a)).ok(); + mon.register(SourceId::from_raw(2), Arc::clone(&cnt_b)).ok(); + mon.check_all(); // baseline + // both unchanged + mon.check_all(); + let stale: Vec<_> = mon.stale_feeds().collect(); + assert_eq!(stale.len(), 2); + } +} diff --git a/crates/transport/src/timer.rs b/crates/transport/src/timer.rs new file mode 100644 index 0000000..cb16565 --- /dev/null +++ b/crates/transport/src/timer.rs @@ -0,0 +1,525 @@ +//! Timer thread — periodic tick and heartbeat event emitter. +//! +//! The timer thread is a simple metronome: it sleeps for a configurable +//! interval, then pushes a [`HotEvent::timer`] with [`TimerKind::Periodic`]. +//! Every N ticks it also emits a [`HotEvent::heartbeat`]. + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread::{self, JoinHandle}; +use std::time::Duration; + +use mantis_events::{EventFlags, HeartbeatPayload, HotEvent, TimerKind, TimerPayload}; +use mantis_types::{InstrumentId, SeqNum, SourceId, Timestamp}; + +use crate::feed::interruptible_sleep; +use crate::tuning::SocketTuning; + +// --------------------------------------------------------------------------- +// Config +// --------------------------------------------------------------------------- + +/// Configuration for the timer thread. +#[derive(Clone, Debug)] +pub struct TimerConfig { + /// Human-readable thread name. + pub name: String, + /// Interval between periodic timer events. + pub tick_interval: Duration, + /// Interval between heartbeat events (must be a multiple of `tick_interval`). + pub heartbeat_interval: Duration, + /// Source ID stamped on every emitted event. + pub source_id: SourceId, + /// Optional CPU core to pin the thread to. + pub core_id: Option, +} + +// --------------------------------------------------------------------------- +// Errors +// --------------------------------------------------------------------------- + +/// Errors from validating a [`TimerConfig`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum TimerConfigError { + /// `tick_interval` was zero. + ZeroTickInterval, + /// `heartbeat_interval` is shorter than `tick_interval`. + HeartbeatShorterThanTick, + /// `heartbeat_interval` is not evenly divisible by `tick_interval`. + HeartbeatNotDivisible, + /// `heartbeat_interval / tick_interval` exceeds `u64::MAX`. + HeartbeatTooLarge, +} + +impl core::fmt::Display for TimerConfigError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::ZeroTickInterval => write!(f, "tick_interval must be > 0"), + Self::HeartbeatShorterThanTick => { + write!(f, "heartbeat_interval must be >= tick_interval") + } + Self::HeartbeatNotDivisible => { + write!( + f, + "heartbeat_interval must be evenly divisible by tick_interval" + ) + } + Self::HeartbeatTooLarge => { + write!(f, "heartbeat_interval / tick_interval exceeds u64::MAX") + } + } + } +} + +impl std::error::Error for TimerConfigError {} + +/// Error returned when spawning the timer thread fails. +#[derive(Debug)] +pub enum TimerSpawnError { + /// The timer configuration is invalid. + Config(TimerConfigError), + /// The OS refused to spawn the thread. + Spawn(std::io::Error), +} + +impl core::fmt::Display for TimerSpawnError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::Config(e) => write!(f, "timer config error: {e}"), + Self::Spawn(e) => write!(f, "timer spawn error: {e}"), + } + } +} + +impl std::error::Error for TimerSpawnError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Config(e) => Some(e), + Self::Spawn(e) => Some(e), + } + } +} + +impl From for TimerSpawnError { + fn from(e: TimerConfigError) -> Self { + Self::Config(e) + } +} + +impl From for TimerSpawnError { + fn from(e: std::io::Error) -> Self { + Self::Spawn(e) + } +} + +// --------------------------------------------------------------------------- +// Validation +// --------------------------------------------------------------------------- + +fn validate(config: &TimerConfig) -> Result { + if config.tick_interval.is_zero() { + return Err(TimerConfigError::ZeroTickInterval); + } + let tick_ns = config.tick_interval.as_nanos(); + let hb_ns = config.heartbeat_interval.as_nanos(); + if hb_ns < tick_ns { + return Err(TimerConfigError::HeartbeatShorterThanTick); + } + if !hb_ns.is_multiple_of(tick_ns) { + return Err(TimerConfigError::HeartbeatNotDivisible); + } + let ratio = hb_ns / tick_ns; + if ratio > u128::from(u64::MAX) { + return Err(TimerConfigError::HeartbeatTooLarge); + } + #[expect(clippy::cast_possible_truncation, reason = "checked above")] + let heartbeat_every = ratio as u64; + Ok(heartbeat_every) +} + +// --------------------------------------------------------------------------- +// Timer thread handle +// --------------------------------------------------------------------------- + +/// Handle to a running timer thread. +/// +/// Dropping the handle signals shutdown (but does not join the thread). +/// Call [`TimerThread::shutdown`] for a graceful stop that waits for exit. +pub struct TimerThread { + join: Option>, + shutdown: Arc, +} + +impl TimerThread { + /// Spawn the timer thread. + /// + /// The `push` callback is invoked on the timer thread for every emitted + /// [`HotEvent`] (both ticks and heartbeats). + /// + /// # Errors + /// + /// Returns [`TimerSpawnError::Config`] if the configuration is invalid, or + /// [`TimerSpawnError::Spawn`] if the OS refuses to create the thread. + pub fn spawn(config: TimerConfig, push: F) -> Result + where + F: FnMut(HotEvent) + Send + 'static, + { + let heartbeat_every = validate(&config)?; + let shutdown = Arc::new(AtomicBool::new(false)); + let shutdown_flag = Arc::clone(&shutdown); + + let join = thread::Builder::new() + .name(config.name.clone()) + .spawn(move || { + timer_loop(&config, heartbeat_every, push, &shutdown_flag); + })?; + + Ok(Self { + join: Some(join), + shutdown, + }) + } + + /// Signal the timer thread to stop and wait for it to exit. + pub fn shutdown(mut self) { + self.shutdown.store(true, Ordering::Release); + if let Some(join) = self.join.take() { + let _ = join.join(); + } + } +} + +impl Drop for TimerThread { + fn drop(&mut self) { + self.shutdown.store(true, Ordering::Release); + } +} + +// --------------------------------------------------------------------------- +// Main loop +// --------------------------------------------------------------------------- + +fn timer_loop(config: &TimerConfig, heartbeat_every: u64, mut push: F, shutdown: &AtomicBool) +where + F: FnMut(HotEvent) + Send + 'static, +{ + let tuning = SocketTuning { + core_id: config.core_id, + #[cfg(feature = "tuning")] + busy_poll_us: None, + }; + tuning.apply_affinity(); + + let mut tick_counter: u64 = 0; + let mut heartbeat_counter: u32 = 0; + let mut seq: u64 = 0; + + loop { + if interruptible_sleep(config.tick_interval, shutdown) { + return; + } + if shutdown.load(Ordering::Acquire) { + return; + } + + tick_counter += 1; + seq += 1; + + let timer_payload = TimerPayload { + timer_id: 0, + kind: TimerKind::Periodic, + _pad: [0; 3], + }; + + push(HotEvent::timer( + Timestamp::now(), + SeqNum::from_raw(seq), + InstrumentId::NONE, + config.source_id, + EventFlags::EMPTY, + timer_payload, + )); + + if tick_counter.is_multiple_of(heartbeat_every) { + heartbeat_counter = heartbeat_counter.wrapping_add(1); + seq += 1; + + push(HotEvent::heartbeat( + Timestamp::now(), + SeqNum::from_raw(seq), + config.source_id, + EventFlags::EMPTY, + HeartbeatPayload { + counter: heartbeat_counter, + }, + )); + } + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use mantis_events::EventBody; + use std::sync::Mutex; + + fn test_source() -> SourceId { + SourceId::from_raw(99) + } + + #[test] + #[expect(clippy::expect_used)] + fn emits_periodic_timer_events() { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let events_push = Arc::clone(&events); + + let timer = TimerThread::spawn( + TimerConfig { + name: "test-timer".into(), + tick_interval: Duration::from_millis(50), + heartbeat_interval: Duration::from_millis(200), + source_id: test_source(), + core_id: None, + }, + move |ev| { + events_push.lock().expect("lock").push(ev); + }, + ) + .expect("spawn"); + + // Generous sleep for CI runners + thread::sleep(Duration::from_millis(1500)); + timer.shutdown(); + + let collected = events.lock().expect("lock"); + let timer_count = collected + .iter() + .filter(|e| matches!(e.body, EventBody::Timer(_))) + .count(); + assert!( + timer_count >= 2, + "expected >= 2 timer events, got {timer_count}" + ); + } + + #[test] + #[expect(clippy::expect_used)] + fn emits_heartbeat_events() { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let events_push = Arc::clone(&events); + + let timer = TimerThread::spawn( + TimerConfig { + name: "test-hb".into(), + tick_interval: Duration::from_millis(50), + heartbeat_interval: Duration::from_millis(200), + source_id: test_source(), + core_id: None, + }, + move |ev| { + events_push.lock().expect("lock").push(ev); + }, + ) + .expect("spawn"); + + // Give generous time for CI runners (slow VMs) + thread::sleep(Duration::from_millis(1500)); + timer.shutdown(); + + let collected = events.lock().expect("lock"); + let hb_count = collected + .iter() + .filter(|e| matches!(e.body, EventBody::Heartbeat(_))) + .count(); + assert!( + hb_count >= 1, + "expected >= 1 heartbeat events, got {hb_count}" + ); + } + + #[test] + #[expect(clippy::expect_used)] + fn seq_is_monotonic() { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let events_push = Arc::clone(&events); + + let timer = TimerThread::spawn( + TimerConfig { + name: "test-seq".into(), + tick_interval: Duration::from_millis(50), + heartbeat_interval: Duration::from_millis(100), + source_id: test_source(), + core_id: None, + }, + move |ev| { + events_push.lock().expect("lock").push(ev); + }, + ) + .expect("spawn"); + + thread::sleep(Duration::from_secs(1)); + timer.shutdown(); + + let collected = events.lock().expect("lock"); + assert!(!collected.is_empty(), "expected some events"); + + for window in collected.windows(2) { + let a = window[0].header.seq; + let b = window[1].header.seq; + assert!(b > a, "seq not strictly increasing: {a:?} -> {b:?}"); + } + } + + #[test] + #[expect(clippy::expect_used)] + fn shutdown_exits_quickly() { + let timer = TimerThread::spawn( + TimerConfig { + name: "test-shutdown".into(), + tick_interval: Duration::from_millis(100), + heartbeat_interval: Duration::from_millis(100), + source_id: test_source(), + core_id: None, + }, + |_| {}, + ) + .expect("spawn"); + + let start = std::time::Instant::now(); + timer.shutdown(); + let elapsed = start.elapsed(); + + assert!( + elapsed < Duration::from_millis(500), + "shutdown took {elapsed:?}, expected < 500ms" + ); + } + + #[test] + fn zero_tick_interval_errors() { + let result = TimerThread::spawn( + TimerConfig { + name: "bad".into(), + tick_interval: Duration::ZERO, + heartbeat_interval: Duration::from_millis(100), + source_id: test_source(), + core_id: None, + }, + |_| {}, + ); + assert!(matches!( + result, + Err(TimerSpawnError::Config(TimerConfigError::ZeroTickInterval)) + )); + } + + #[test] + fn heartbeat_shorter_than_tick_errors() { + let result = TimerThread::spawn( + TimerConfig { + name: "bad".into(), + tick_interval: Duration::from_millis(100), + heartbeat_interval: Duration::from_millis(50), + source_id: test_source(), + core_id: None, + }, + |_| {}, + ); + assert!(matches!( + result, + Err(TimerSpawnError::Config( + TimerConfigError::HeartbeatShorterThanTick + )) + )); + } + + #[test] + fn heartbeat_not_divisible_errors() { + let result = TimerThread::spawn( + TimerConfig { + name: "bad".into(), + tick_interval: Duration::from_millis(30), + heartbeat_interval: Duration::from_millis(100), + source_id: test_source(), + core_id: None, + }, + |_| {}, + ); + assert!(matches!( + result, + Err(TimerSpawnError::Config( + TimerConfigError::HeartbeatNotDivisible + )) + )); + } + + #[test] + fn timer_config_error_display() { + let e = TimerConfigError::ZeroTickInterval; + assert!(!e.to_string().is_empty()); + let e = TimerConfigError::HeartbeatShorterThanTick; + assert!(!e.to_string().is_empty()); + let e = TimerConfigError::HeartbeatNotDivisible; + assert!(!e.to_string().is_empty()); + let e = TimerConfigError::HeartbeatTooLarge; + assert!(!e.to_string().is_empty()); + } + + #[test] + fn timer_config_error_is_std_error() { + let e: &dyn std::error::Error = &TimerConfigError::ZeroTickInterval; + assert!(e.source().is_none()); + } + + #[test] + fn timer_spawn_error_display() { + let e = TimerSpawnError::Config(TimerConfigError::ZeroTickInterval); + assert!(!e.to_string().is_empty()); + // Also cover the Spawn variant display + let io_err = std::io::Error::other("test"); + let e = TimerSpawnError::Spawn(io_err); + assert!(!e.to_string().is_empty()); + } + + #[test] + fn timer_spawn_error_source() { + let e = TimerSpawnError::Config(TimerConfigError::ZeroTickInterval); + assert!(std::error::Error::source(&e).is_some()); + + let io_err = std::io::Error::other("test"); + let e = TimerSpawnError::Spawn(io_err); + assert!(std::error::Error::source(&e).is_some()); + } + + #[test] + #[expect(clippy::expect_used)] + fn timer_drop_signals_shutdown() { + let config = TimerConfig { + name: "test-drop".into(), + tick_interval: Duration::from_millis(50), + heartbeat_interval: Duration::from_millis(100), + source_id: test_source(), + core_id: None, + }; + let timer = TimerThread::spawn(config, |_| {}).expect("spawn"); + drop(timer); + // If we get here, Drop didn't hang — test passes + } + + #[test] + fn heartbeat_too_large_errors() { + let config = TimerConfig { + name: "test-too-large".into(), + tick_interval: Duration::from_nanos(1), + heartbeat_interval: Duration::from_secs(u64::MAX), + source_id: test_source(), + core_id: None, + }; + let result = TimerThread::spawn(config, |_| {}); + assert!(result.is_err()); + } +} diff --git a/crates/types/src/timestamp.rs b/crates/types/src/timestamp.rs index 910da97..7dc5978 100644 --- a/crates/types/src/timestamp.rs +++ b/crates/types/src/timestamp.rs @@ -22,6 +22,30 @@ impl Timestamp { pub const fn as_nanos(self) -> u64 { self.0 } + + /// Returns the current time as nanoseconds since the Unix epoch. + /// + /// Uses `std::time::SystemTime` for wall-clock time. + /// + /// # Panics + /// + /// Panics if the system clock is before the Unix epoch. + #[cfg(feature = "std")] + #[must_use] + #[expect( + clippy::expect_used, + reason = "system clock before epoch is unrecoverable" + )] + #[expect( + clippy::cast_possible_truncation, + reason = "u64 nanos covers ~584 years from epoch — sufficient until 2554" + )] + pub fn now() -> Self { + let duration = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock is before Unix epoch"); + Self(duration.as_nanos() as u64) + } } impl fmt::Debug for Timestamp { @@ -68,4 +92,19 @@ mod tests { assert_eq!(ts.to_string(), "42ns"); assert_eq!(alloc::format!("{ts:?}"), "Timestamp(42ns)"); } + + #[cfg(feature = "std")] + #[test] + fn now_returns_nonzero() { + let ts = Timestamp::now(); + assert!(ts.as_nanos() > 0); + } + + #[cfg(feature = "std")] + #[test] + fn now_is_non_decreasing() { + let a = Timestamp::now(); + let b = Timestamp::now(); + assert!(b.as_nanos() >= a.as_nanos()); + } } diff --git a/crates/verify/Cargo.toml b/crates/verify/Cargo.toml index 3dfffdb..1789ca7 100644 --- a/crates/verify/Cargo.toml +++ b/crates/verify/Cargo.toml @@ -15,6 +15,9 @@ mantis-fixed = { workspace = true } mantis-seqlock = { workspace = true } mantis-events = { workspace = true } mantis-market-state = { workspace = true } +mantis-binance = { workspace = true } +mantis-polymarket = { workspace = true } +mantis-registry = { workspace = true } bolero = { workspace = true } [lints] diff --git a/crates/verify/src/decoder_props.rs b/crates/verify/src/decoder_props.rs new file mode 100644 index 0000000..789e553 --- /dev/null +++ b/crates/verify/src/decoder_props.rs @@ -0,0 +1,105 @@ +//! Bolero property tests for venue decoders. +//! +//! Ensures that both the Binance and Polymarket decoders never panic +//! on arbitrary input — critical for HFT where a panic on malformed +//! data crashes the trading system. + +#[cfg(test)] +mod tests { + use bolero::check; + use mantis_events::{EventFlags, HeartbeatPayload, HotEvent}; + use mantis_fixed::FixedI64; + use mantis_types::{InstrumentId, InstrumentMeta, SeqNum, SourceId, Timestamp}; + + fn make_out() -> [HotEvent; 64] { + [HotEvent::heartbeat( + Timestamp::ZERO, + SeqNum::ZERO, + SourceId::from_raw(0), + EventFlags::EMPTY, + HeartbeatPayload { counter: 0 }, + ); 64] + } + + // ---- Binance decoder property tests ---- + + #[test] + #[expect( + clippy::expect_used, + reason = "test-only setup with known-valid constants" + )] + fn binance_decoder_never_panics() { + use mantis_binance::{BinanceDecoder, BinanceSymbolMapping}; + + let meta = InstrumentMeta::new( + FixedI64::<3>::from_str_decimal("0.01").expect("valid tick_size"), + FixedI64::<3>::from_str_decimal("0.001").expect("valid lot_size"), + ) + .expect("valid meta"); + + let mut decoder = BinanceDecoder::new( + SourceId::from_raw(2), + &[BinanceSymbolMapping { + symbol: "BTCUSDT", + instrument_id: InstrumentId::from_raw(1), + meta, + }], + ) + .expect("valid decoder"); + + check!().for_each(|bytes: &[u8]| { + let mut buf = bytes.to_vec(); + let mut out = make_out(); + let _ = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + // Must not panic — any return value is acceptable + }); + } + + // ---- Polymarket decoder property tests ---- + + #[test] + #[expect( + clippy::expect_used, + reason = "test-only setup with known-valid constants" + )] + fn polymarket_decoder_never_panics() { + use mantis_polymarket::market::PolymarketMarketDecoder; + use mantis_registry::{ + Asset, InstrumentKey, OutcomeSide, PolymarketBinding, PolymarketWindowBinding, + Timeframe, + }; + + let mut reg = mantis_registry::InstrumentRegistry::<6>::new(); + let meta = InstrumentMeta::new( + FixedI64::<6>::from_raw(10_000), + FixedI64::<6>::from_raw(1_000_000), + ) + .expect("valid meta"); + + let key = InstrumentKey::prediction(Asset::Btc, Timeframe::M15, OutcomeSide::Up); + let id = reg + .insert(key, meta, None, Some(PolymarketBinding::default())) + .expect("insert ok"); + + reg.bind_polymarket_current( + id, + PolymarketWindowBinding { + token_id: "abc123".to_owned(), + market_slug: "btc-15m-up".to_owned(), + window_start: Timestamp::from_nanos(0), + window_end: Timestamp::from_nanos(900_000_000_000), + condition_id: None, + }, + ) + .expect("bind ok"); + + let mut decoder = PolymarketMarketDecoder::<6>::new(SourceId::from_raw(10), ®); + + check!().for_each(|bytes: &[u8]| { + let mut buf = bytes.to_vec(); + let mut out = make_out(); + let _ = decoder.decode(&mut buf, Timestamp::from_nanos(0), &mut out); + // Must not panic — any return value is acceptable + }); + } +} diff --git a/crates/verify/src/lib.rs b/crates/verify/src/lib.rs index 3ce7723..bdcd947 100644 --- a/crates/verify/src/lib.rs +++ b/crates/verify/src/lib.rs @@ -5,6 +5,7 @@ #![deny(unsafe_code)] +mod decoder_props; mod fixed_props; pub mod market_state_props; pub mod seqlock_proofs; diff --git a/docs/PROGRESS.md b/docs/PROGRESS.md index 93ad755..fc351b6 100644 --- a/docs/PROGRESS.md +++ b/docs/PROGRESS.md @@ -176,6 +176,30 @@ - [x] Old `Strategy` trait removed from `mantis-market-state` (clean break, no deprecation) - [x] 38 unit tests, `no_std` clean, clippy clean, 3 rounds of Codex review +### 1.13 Transport & Venue Decoders (`mantis-transport`, `mantis-binance`, `mantis-polymarket`) +**Status: Complete** | Completed: 2026-04-11 + +- [x] WebSocket transport layer — blocking IO on CPU-pinned threads, no async runtime +- [x] Polymarket market WS adapter — heartbeat, subscription, reconnect with backoff +- [x] Binance futures bookTicker adapter — single + multi-symbol URL generation +- [x] `FnMut(&mut [u8]) -> bool` callback for simd-json in-place parsing +- [x] `BinanceDecoder` — multi-symbol (1-8) with flat lookup, combined stream peek +- [x] `PolymarketMarketDecoder<'r, D>` — book/price_change/trade with peek_type dispatch +- [x] Zero-alloc decode path — borrowed serde structs, `parse_decimal_bytes`, Ticks/Lots conversion +- [x] simd-json pluggable via feature flag (serde_json fallback) +- [x] `spawn_binance_feed` / `spawn_polymarket_market_feed` — monomorphized push callback +- [x] Backpressure: `FnMut(HotEvent) -> bool` with drop_count tracking +- [x] `FeedSpawnResult` with event_count + drop_count for monitoring +- [x] `TimerThread` — periodic Timer(Periodic) + Heartbeat events, validated config +- [x] `FeedMonitor` — stale feed detection via event_count comparison, first-check baseline +- [x] `Timestamp::now()` — wall-clock access (std feature) +- [x] Truncation warning for Polymarket snapshots >64 levels +- [x] ~307ns Binance bookTicker, ~306ns Polymarket price_change decode latency +- [x] Criterion benchmarks (decode.rs) + decimal parse contenders (fixed.rs) +- [x] 13 binance tests, 21 polymarket tests, 7 timer tests, 8 monitor tests +- [x] Bolero property tests: arbitrary bytes → decoders never panic +- [x] READMEs for all three crates + ### 1.4 Snapshot Publication - [ ] Single-writer publication primitive - [ ] Lock-free reader access @@ -251,3 +275,7 @@ | `mantis-layout` | Active | std | 6 | — | — | | `mantis-verify` | Active | std | 13 | — | 4 kani proofs, 13 bolero/diff | | `mantis-strategy` | Active | yes | 38 | — | 3 rounds Codex review | +| `mantis-transport` | Active | std | 22 | — | — | +| `mantis-binance` | Active | std | 13 | 1 criterion group | 1 bolero prop | +| `mantis-polymarket` | Active | std | 21 | 4 criterion groups | 1 bolero prop | +| `mantis-registry` | Active | std | 41 | — | — |