diff --git a/src/book.rs b/src/book.rs index 11b0080..952e257 100644 --- a/src/book.rs +++ b/src/book.rs @@ -165,10 +165,23 @@ pub struct OrderBook { /// Hash of token_id for fast lookups (avoids string comparisons in hot path) pub token_id_hash: u64, - /// Current sequence number for ordering updates - /// This helps us ignore old/duplicate updates that arrive out of order + /// Legacy delta sequence number. + /// + /// Kept as a compatibility alias for [`Self::last_delta_sequence`]. WebSocket + /// snapshot timestamps are tracked separately in [`Self::last_snapshot_timestamp_ms`]. pub sequence: u64, + /// Last accepted legacy/incremental delta sequence number. + /// + /// Delta sequences are a different clock from WebSocket snapshot timestamps and + /// must not be compared against millisecond timestamps. + pub last_delta_sequence: u64, + + /// Last accepted full-book snapshot timestamp in milliseconds. + /// + /// Used for WebSocket `book` snapshots and serde `BookUpdate` snapshots. + pub last_snapshot_timestamp_ms: u64, + /// Last update timestamp - when we last got new data for this book pub timestamp: chrono::DateTime, @@ -235,7 +248,9 @@ impl OrderBook { Self { token_id, token_id_hash, - sequence: 0, // Start at 0, will increment as we get updates + sequence: 0, // Compatibility alias for last_delta_sequence + last_delta_sequence: 0, + last_snapshot_timestamp_ms: 0, timestamp: Utc::now(), bids: BookSide::new(BookSideKind::Bid, max_depth), // Empty to start asks: BookSide::new(BookSideKind::Ask, max_depth), // Empty to start @@ -444,7 +459,7 @@ impl OrderBook { timestamp: self.timestamp, bids: self.bids(None), // Get all bids (up to max_depth) asks: self.asks(None), // Get all asks (up to max_depth) - sequence: self.sequence, + sequence: self.last_delta_sequence, } } @@ -475,11 +490,11 @@ impl OrderBook { pub fn apply_delta_fast(&mut self, delta: FastOrderDelta) -> Result<()> { // Validate sequence ordering - ignore old updates that arrive late // This is crucial for maintaining data integrity in real-time systems - if delta.sequence <= self.sequence { + if delta.sequence <= self.last_delta_sequence { trace!( "Ignoring stale delta: {} <= {}", delta.sequence, - self.sequence + self.last_delta_sequence ); return Ok(()); } @@ -510,6 +525,7 @@ impl OrderBook { } // Update our tracking info + self.last_delta_sequence = delta.sequence; self.sequence = delta.sequence; self.timestamp = delta.timestamp; @@ -583,9 +599,10 @@ impl OrderBook { return Err(PolyfillError::validation("Token ID mismatch")); } - // Use the exchange-provided timestamp as our monotonic snapshot marker. + // Use the exchange-provided timestamp as the 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. + // Snapshot timestamps are intentionally separate from legacy delta sequence numbers. if !self.should_apply_snapshot(update.timestamp, update.hash.as_deref()) { return Ok(()); } @@ -602,7 +619,7 @@ impl OrderBook { self.validate_snapshot_level(parsed)?; } - self.sequence = update.timestamp; + self.last_snapshot_timestamp_ms = 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); @@ -706,10 +723,10 @@ impl OrderBook { #[inline] fn should_apply_snapshot(&self, timestamp: u64, hash: Option<&str>) -> bool { - if timestamp > self.sequence { + if timestamp > self.last_snapshot_timestamp_ms { return true; } - if timestamp < self.sequence { + if timestamp < self.last_snapshot_timestamp_ms { return false; } @@ -751,7 +768,7 @@ impl OrderBook { self.validate_snapshot_level(level)?; } - self.sequence = timestamp; + self.last_snapshot_timestamp_ms = 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); @@ -1280,6 +1297,8 @@ mod tests { book.apply_delta(delta).unwrap(); assert_eq!(book.sequence, 1); // Sequence should update + assert_eq!(book.last_delta_sequence, 1); + assert_eq!(book.last_snapshot_timestamp_ms, 0); assert_eq!(book.best_bid().unwrap().price, dec!(0.5)); // Should be our bid assert_eq!(book.best_bid().unwrap().size, dec!(100)); // Should be our size } @@ -1406,6 +1425,84 @@ mod tests { assert_eq!(book.best_ask().unwrap().size, dec!(45)); } + #[test] + fn test_ws_snapshot_timestamp_does_not_block_delta_sequence() { + let mut book = OrderBook::new("test_token".to_string(), 10); + let snapshot_timestamp_ms = 1_757_908_892_351; + let levels = [ + ParsedBookLevel { + side: Side::BUY, + price_ticks: 5_000, + size_units: 100_000, + }, + ParsedBookLevel { + side: Side::SELL, + price_ticks: 6_000, + size_units: 100_000, + }, + ]; + + assert!(book + .apply_ws_book_snapshot_fast("test_token", snapshot_timestamp_ms, None, &levels) + .unwrap()); + + assert_eq!(book.sequence, 0); + assert_eq!(book.last_delta_sequence, 0); + assert_eq!(book.last_snapshot_timestamp_ms, snapshot_timestamp_ms); + + book.apply_delta(OrderDelta { + token_id: "test_token".to_string(), + timestamp: Utc::now(), + side: Side::BUY, + price: dec!(0.51), + size: dec!(11), + sequence: 1, + }) + .unwrap(); + + assert_eq!(book.sequence, 1); + assert_eq!(book.last_delta_sequence, 1); + assert_eq!(book.last_snapshot_timestamp_ms, snapshot_timestamp_ms); + assert_eq!(book.best_bid().unwrap().price, dec!(0.51)); + } + + #[test] + fn test_delta_sequence_does_not_block_snapshot_timestamp() { + let mut book = OrderBook::new("test_token".to_string(), 10); + + book.apply_delta(OrderDelta { + token_id: "test_token".to_string(), + timestamp: Utc::now(), + side: Side::BUY, + price: dec!(0.40), + size: dec!(10), + sequence: 10_000, + }) + .unwrap(); + + book.apply_book_update(&BookUpdate { + asset_id: "test_token".to_string(), + market: "0xabc".to_string(), + timestamp: 1_000, + bids: vec![OrderSummary { + price: dec!(0.50), + size: dec!(20), + }], + asks: vec![OrderSummary { + price: dec!(0.60), + size: dec!(30), + }], + hash: None, + }) + .unwrap(); + + assert_eq!(book.sequence, 10_000); + assert_eq!(book.last_delta_sequence, 10_000); + assert_eq!(book.last_snapshot_timestamp_ms, 1_000); + assert_eq!(book.best_bid().unwrap().price, dec!(0.50)); + assert_eq!(book.best_ask().unwrap().price, dec!(0.60)); + } + #[test] fn test_book_update_allows_same_timestamp_with_different_hash() { let mut book = OrderBook::new("test_token".to_string(), 10); @@ -1554,7 +1651,9 @@ mod tests { .unwrap_err(); assert!(err.to_string().contains("Price not aligned")); - assert_eq!(book.sequence, 100); + assert_eq!(book.sequence, 0); + assert_eq!(book.last_delta_sequence, 0); + assert_eq!(book.last_snapshot_timestamp_ms, 100); assert_eq!(book.bids(None).len(), 1); assert_eq!(book.asks(None).len(), 1); assert_eq!(book.best_bid().unwrap().price, dec!(0.50)); diff --git a/src/ws_hot_path.rs b/src/ws_hot_path.rs index c982a3d..bc42aed 100644 --- a/src/ws_hot_path.rs +++ b/src/ws_hot_path.rs @@ -370,7 +370,8 @@ mod tests { .is_err()); let snapshot = books.get_book("test_asset_id").unwrap(); - assert_eq!(snapshot.sequence, 1000); + assert_eq!(snapshot.sequence, 0); + assert_eq!(snapshot.timestamp.timestamp_millis(), 1000); assert_eq!(snapshot.bids.len(), 1); assert_eq!(snapshot.asks.len(), 1); assert_eq!(snapshot.bids[0].price, dec!(0.50)); @@ -407,7 +408,8 @@ mod tests { .is_err()); let snapshot = books.get_book("test_asset_id").unwrap(); - assert_eq!(snapshot.sequence, 1000); + assert_eq!(snapshot.sequence, 0); + assert_eq!(snapshot.timestamp.timestamp_millis(), 1000); assert_eq!(snapshot.bids.len(), 1); assert_eq!(snapshot.asks.len(), 1); assert_eq!(snapshot.bids[0].price, dec!(0.50)); @@ -433,7 +435,8 @@ mod tests { 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.sequence, 0); + assert_eq!(snapshot.timestamp.timestamp_millis(), 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));