diff --git a/src/book.rs b/src/book.rs index 81ed0a4..11b0080 100644 --- a/src/book.rs +++ b/src/book.rs @@ -195,6 +195,12 @@ pub struct OrderBook { /// Snapshot generation used to retain book levels without rescanning input payloads. snapshot_generation: u64, + /// Last accepted full-book snapshot hash, when the feed provides one. + /// + /// This is used as a tie-breaker for same-millisecond snapshots. If the timestamp is equal + /// but the hash differs, the snapshot can still represent a newer book state. + last_snapshot_hash: Option, + /// Minimum tick size for this market in ticks (like 10 for $0.001 increments) /// Some markets only allow certain price increments /// We store this in ticks for fast validation without conversion @@ -234,6 +240,7 @@ impl OrderBook { bids: BookSide::new(BookSideKind::Bid, max_depth), // Empty to start asks: BookSide::new(BookSideKind::Ask, max_depth), // Empty to start snapshot_generation: 0, + last_snapshot_hash: None, tick_size_ticks: None, // We'll set this later when we learn about the market max_depth, } @@ -531,16 +538,13 @@ impl OrderBook { &self, asset_id: &str, timestamp: u64, + hash: Option<&str>, ) -> Result { if asset_id != self.token_id { return Err(PolyfillError::validation("Token ID mismatch")); } - if timestamp <= self.sequence { - return Ok(false); - } - - Ok(true) + Ok(self.should_apply_snapshot(timestamp, hash)) } /// Atomically apply a WebSocket `book` snapshot. @@ -552,13 +556,14 @@ impl OrderBook { &mut self, asset_id: &str, timestamp: u64, + hash: Option<&str>, levels: &[ParsedBookLevel], ) -> Result { - if !self.should_apply_ws_book_update(asset_id, timestamp)? { + if !self.should_apply_ws_book_update(asset_id, timestamp, hash)? { return Ok(false); } - self.apply_validated_snapshot(timestamp, levels)?; + self.apply_validated_snapshot(timestamp, hash, levels)?; Ok(true) } @@ -578,10 +583,10 @@ impl OrderBook { return Err(PolyfillError::validation("Token ID mismatch")); } - // Use the exchange-provided timestamp as our monotonic sequence marker. - // This is less strict than the REST/legacy delta sequence but works for - // ignoring obviously stale book snapshots. - if update.timestamp <= self.sequence { + // Use the exchange-provided timestamp as our monotonic snapshot marker. + // If the feed emits two snapshots in the same millisecond, a different hash is treated as + // a distinct newer state. Equal timestamp without a hash is considered duplicate/stale. + if !self.should_apply_snapshot(update.timestamp, update.hash.as_deref()) { return Ok(()); } @@ -598,6 +603,7 @@ impl OrderBook { } self.sequence = update.timestamp; + self.last_snapshot_hash = update.hash.clone(); self.timestamp = chrono::DateTime::::from_timestamp_millis(update.timestamp as i64) .unwrap_or_else(Utc::now); self.begin_snapshot(); @@ -698,6 +704,18 @@ impl OrderBook { self.snapshot_generation = self.snapshot_generation.wrapping_add(1); } + #[inline] + fn should_apply_snapshot(&self, timestamp: u64, hash: Option<&str>) -> bool { + if timestamp > self.sequence { + return true; + } + if timestamp < self.sequence { + return false; + } + + hash.is_some_and(|hash| self.last_snapshot_hash.as_deref() != Some(hash)) + } + #[inline] fn parse_snapshot_summary(&self, side: Side, level: &OrderSummary) -> Result { let price_ticks = decimal_to_price(level.price) @@ -726,6 +744,7 @@ impl OrderBook { fn apply_validated_snapshot( &mut self, timestamp: u64, + hash: Option<&str>, levels: &[ParsedBookLevel], ) -> Result<()> { for &level in levels { @@ -733,6 +752,7 @@ impl OrderBook { } self.sequence = timestamp; + self.last_snapshot_hash = hash.map(str::to_owned); self.timestamp = chrono::DateTime::::from_timestamp_millis(timestamp as i64) .unwrap_or_else(Utc::now); self.begin_snapshot(); @@ -1386,6 +1406,109 @@ mod tests { assert_eq!(book.best_ask().unwrap().size, dec!(45)); } + #[test] + fn test_book_update_allows_same_timestamp_with_different_hash() { + let mut book = OrderBook::new("test_token".to_string(), 10); + let timestamp = 1_757_908_892_351; + + book.apply_book_update(&BookUpdate { + asset_id: "test_token".to_string(), + market: "0xabc".to_string(), + timestamp, + bids: vec![OrderSummary { + price: dec!(0.48), + size: dec!(10), + }], + asks: vec![OrderSummary { + price: dec!(0.52), + size: dec!(20), + }], + hash: Some("hash_a".to_string()), + }) + .unwrap(); + + book.apply_book_update(&BookUpdate { + asset_id: "test_token".to_string(), + market: "0xabc".to_string(), + timestamp, + bids: vec![OrderSummary { + price: dec!(0.49), + size: dec!(30), + }], + asks: vec![OrderSummary { + price: dec!(0.53), + size: dec!(40), + }], + hash: Some("hash_b".to_string()), + }) + .unwrap(); + + assert_eq!(book.best_bid().unwrap().price, dec!(0.49)); + assert_eq!(book.best_bid().unwrap().size, dec!(30)); + assert_eq!(book.best_ask().unwrap().price, dec!(0.53)); + assert_eq!(book.best_ask().unwrap().size, dec!(40)); + + book.apply_book_update(&BookUpdate { + asset_id: "test_token".to_string(), + market: "0xabc".to_string(), + timestamp, + bids: vec![OrderSummary { + price: dec!(0.50), + size: dec!(50), + }], + asks: vec![OrderSummary { + price: dec!(0.54), + size: dec!(60), + }], + hash: Some("hash_b".to_string()), + }) + .unwrap(); + + assert_eq!(book.best_bid().unwrap().price, dec!(0.49)); + assert_eq!(book.best_ask().unwrap().price, dec!(0.53)); + } + + #[test] + fn test_book_update_rejects_same_timestamp_without_hash_tie_breaker() { + let mut book = OrderBook::new("test_token".to_string(), 10); + let timestamp = 1_757_908_892_351; + + book.apply_book_update(&BookUpdate { + asset_id: "test_token".to_string(), + market: "0xabc".to_string(), + timestamp, + bids: vec![OrderSummary { + price: dec!(0.48), + size: dec!(10), + }], + asks: vec![OrderSummary { + price: dec!(0.52), + size: dec!(20), + }], + hash: None, + }) + .unwrap(); + + book.apply_book_update(&BookUpdate { + asset_id: "test_token".to_string(), + market: "0xabc".to_string(), + timestamp, + bids: vec![OrderSummary { + price: dec!(0.49), + size: dec!(30), + }], + asks: vec![OrderSummary { + price: dec!(0.53), + size: dec!(40), + }], + hash: None, + }) + .unwrap(); + + assert_eq!(book.best_bid().unwrap().price, dec!(0.48)); + assert_eq!(book.best_ask().unwrap().price, dec!(0.52)); + } + #[test] fn test_book_update_error_keeps_existing_snapshot() { let mut book = OrderBook::new("test_token".to_string(), 10); diff --git a/src/ws_hot_path.rs b/src/ws_hot_path.rs index 3e7747f..5951d11 100644 --- a/src/ws_hot_path.rs +++ b/src/ws_hot_path.rs @@ -130,6 +130,7 @@ fn process_stream_object<'tape, 'input>( .ok_or_else(|| PolyfillError::parse("Missing timestamp", None))?; let timestamp = parse_u64(timestamp_value) .ok_or_else(|| PolyfillError::parse("Invalid timestamp", None))?; + let hash = obj.get("hash").and_then(|v| v.into_string()); let bids = obj .get("bids") @@ -145,7 +146,7 @@ fn process_stream_object<'tape, 'input>( let result = books.with_book_mut(asset_id, |book| { parsed_levels.clear(); - if !book.should_apply_ws_book_update(asset_id, timestamp)? { + if !book.should_apply_ws_book_update(asset_id, timestamp, hash)? { return Ok(0); } @@ -153,7 +154,7 @@ fn process_stream_object<'tape, 'input>( collect_levels(Side::SELL, asks, parsed_levels)?; let parsed_count = parsed_levels.len(); - if book.apply_ws_book_snapshot_fast(asset_id, timestamp, parsed_levels)? { + if book.apply_ws_book_snapshot_fast(asset_id, timestamp, hash, parsed_levels)? { Ok(parsed_count) } else { Ok(0) @@ -411,4 +412,40 @@ mod tests { assert_eq!(snapshot.bids[0].price, dec!(0.50)); assert_eq!(snapshot.asks[0].price, dec!(0.60)); } + + #[test] + fn processor_allows_same_timestamp_with_different_hash() { + let books = OrderBookManager::new(10); + books.get_or_create_book("test_asset_id").unwrap(); + let mut processor = WsBookUpdateProcessor::new(1024); + + let mut first = br#"{"event_type":"book","asset_id":"test_asset_id","market":"0xabc","timestamp":1000,"hash":"hash_a","bids":[{"price":"0.5000","size":"10.0000"}],"asks":[{"price":"0.6000","size":"20.0000"}]}"#.to_vec(); + let first_stats = processor + .process_bytes(first.as_mut_slice(), &books) + .unwrap(); + assert_eq!(first_stats.book_levels_applied, 2); + + let mut second = br#"{"event_type":"book","asset_id":"test_asset_id","market":"0xabc","timestamp":1000,"hash":"hash_b","bids":[{"price":"0.5100","size":"11.0000"}],"asks":[{"price":"0.6100","size":"21.0000"}]}"#.to_vec(); + let second_stats = processor + .process_bytes(second.as_mut_slice(), &books) + .unwrap(); + assert_eq!(second_stats.book_levels_applied, 2); + + let snapshot = books.get_book("test_asset_id").unwrap(); + assert_eq!(snapshot.sequence, 1000); + assert_eq!(snapshot.bids[0].price, dec!(0.51)); + assert_eq!(snapshot.bids[0].size, dec!(11)); + assert_eq!(snapshot.asks[0].price, dec!(0.61)); + assert_eq!(snapshot.asks[0].size, dec!(21)); + + let mut duplicate = br#"{"event_type":"book","asset_id":"test_asset_id","market":"0xabc","timestamp":1000,"hash":"hash_b","bids":[{"price":"0.5200","size":"12.0000"}],"asks":[{"price":"0.6200","size":"22.0000"}]}"#.to_vec(); + let duplicate_stats = processor + .process_bytes(duplicate.as_mut_slice(), &books) + .unwrap(); + assert_eq!(duplicate_stats.book_levels_applied, 0); + + let snapshot = books.get_book("test_asset_id").unwrap(); + assert_eq!(snapshot.bids[0].price, dec!(0.51)); + assert_eq!(snapshot.asks[0].price, dec!(0.61)); + } }