mirror of
https://github.com/floor-licker/polyfill-rs.git
synced 2026-08-22 08:58:12 +00:00
Merge pull request #69 from floor-licker/fix/public-book-snapshot-clocks
fix: expose book snapshot clocks
This commit is contained in:
@@ -78,6 +78,8 @@ Run the WS hot-path benchmark locally with `cargo bench --bench ws_hot_path`.
|
|||||||
|
|
||||||
**Parsing paths:** `polyfill-rs` keeps two parsing layers on purpose. The allocation-sensitive WS `book` path uses `WsBookUpdateProcessor` in `src/ws_hot_path.rs`, which walks a reusable `simd-json` tape and applies fixed-point book levels directly. The generic stream parser in `src/decode.rs` is an ergonomic compatibility path: it parses through `serde_json::Value` so it can tolerate batches, unknown event types, and mixed message shapes. Likewise, several generic numeric/decimal deserializers in `src/decode.rs` accept string-or-number API fields through `serde_json::Value`; they are not the zero-allocation hot path.
|
**Parsing paths:** `polyfill-rs` keeps two parsing layers on purpose. The allocation-sensitive WS `book` path uses `WsBookUpdateProcessor` in `src/ws_hot_path.rs`, which walks a reusable `simd-json` tape and applies fixed-point book levels directly. The generic stream parser in `src/decode.rs` is an ergonomic compatibility path: it parses through `serde_json::Value` so it can tolerate batches, unknown event types, and mixed message shapes. Likewise, several generic numeric/decimal deserializers in `src/decode.rs` accept string-or-number API fields through `serde_json::Value`; they are not the zero-allocation hot path.
|
||||||
|
|
||||||
|
**WebSocket snapshot ordering:** Polymarket `book` messages expose a millisecond `timestamp` and optional `hash`, but no monotonic server sequence/version. The book applier treats newer timestamps as newer, rejects older timestamps, suppresses exact same-timestamp/same-hash duplicates, and accepts same-timestamp/different-hash snapshots in websocket arrival order. The hash distinguishes duplicate vs distinct state; it is not a logical ordering key.
|
||||||
|
|
||||||
**Key Performance Optimizations:**
|
**Key Performance Optimizations:**
|
||||||
|
|
||||||
The 21.4% performance improvement comes from HTTP/2 tuning with 512KB stream windows optimized for 469KB payloads, explicit Polymarket request headers, SIMD-backed parsing where the client uses the typed fast-response helper for large REST market responses, and opt-in connection prewarming/keep-alive support.
|
The 21.4% performance improvement comes from HTTP/2 tuning with 512KB stream windows optimized for 469KB payloads, explicit Polymarket request headers, SIMD-backed parsing where the client uses the typed fast-response helper for large REST market responses, and opt-in connection prewarming/keep-alive support.
|
||||||
|
|||||||
+13
-3
@@ -460,6 +460,8 @@ impl OrderBook {
|
|||||||
bids: self.bids(None), // Get all bids (up to max_depth)
|
bids: self.bids(None), // Get all bids (up to max_depth)
|
||||||
asks: self.asks(None), // Get all asks (up to max_depth)
|
asks: self.asks(None), // Get all asks (up to max_depth)
|
||||||
sequence: self.last_delta_sequence,
|
sequence: self.last_delta_sequence,
|
||||||
|
last_delta_sequence: self.last_delta_sequence,
|
||||||
|
last_snapshot_timestamp_ms: self.last_snapshot_timestamp_ms,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -599,9 +601,10 @@ impl OrderBook {
|
|||||||
return Err(PolyfillError::validation("Token ID mismatch"));
|
return Err(PolyfillError::validation("Token ID mismatch"));
|
||||||
}
|
}
|
||||||
|
|
||||||
// Use the exchange-provided timestamp as the monotonic snapshot marker.
|
// Use the exchange-provided timestamp as the primary snapshot marker. The WS `book`
|
||||||
// If the feed emits two snapshots in the same millisecond, a different hash is treated as
|
// message does not expose a monotonic server sequence/version, so same-millisecond
|
||||||
// a distinct newer state. Equal timestamp without a hash is considered duplicate/stale.
|
// snapshots with different hashes are ordered by websocket arrival order. The hash
|
||||||
|
// distinguishes duplicate vs distinct state; it is not a logical ordering key.
|
||||||
// Snapshot timestamps are intentionally separate from legacy delta sequence numbers.
|
// Snapshot timestamps are intentionally separate from legacy delta sequence numbers.
|
||||||
if !self.should_apply_snapshot(update.timestamp, update.hash.as_deref()) {
|
if !self.should_apply_snapshot(update.timestamp, update.hash.as_deref()) {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
@@ -723,6 +726,8 @@ impl OrderBook {
|
|||||||
|
|
||||||
#[inline]
|
#[inline]
|
||||||
fn should_apply_snapshot(&self, timestamp: u64, hash: Option<&str>) -> bool {
|
fn should_apply_snapshot(&self, timestamp: u64, hash: Option<&str>) -> bool {
|
||||||
|
// Without a server sequence/version, equal-timestamp distinct hashes can only be ordered
|
||||||
|
// by arrival. Older timestamps are always stale; exact same timestamp/hash is duplicate.
|
||||||
if timestamp > self.last_snapshot_timestamp_ms {
|
if timestamp > self.last_snapshot_timestamp_ms {
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
@@ -1526,6 +1531,11 @@ mod tests {
|
|||||||
assert_eq!(book.last_snapshot_timestamp_ms, 1_000);
|
assert_eq!(book.last_snapshot_timestamp_ms, 1_000);
|
||||||
assert_eq!(book.best_bid().unwrap().price, dec!(0.50));
|
assert_eq!(book.best_bid().unwrap().price, dec!(0.50));
|
||||||
assert_eq!(book.best_ask().unwrap().price, dec!(0.60));
|
assert_eq!(book.best_ask().unwrap().price, dec!(0.60));
|
||||||
|
|
||||||
|
let snapshot = book.snapshot();
|
||||||
|
assert_eq!(snapshot.sequence, 10_000);
|
||||||
|
assert_eq!(snapshot.last_delta_sequence, 10_000);
|
||||||
|
assert_eq!(snapshot.last_snapshot_timestamp_ms, 1_000);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
@@ -341,6 +341,8 @@ impl Decoder<OrderBook> for RawOrderBookResponse {
|
|||||||
bids,
|
bids,
|
||||||
asks,
|
asks,
|
||||||
sequence: 0, // TODO: Get from response if available
|
sequence: 0, // TODO: Get from response if available
|
||||||
|
last_delta_sequence: 0,
|
||||||
|
last_snapshot_timestamp_ms: 0,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+67
-2
@@ -358,7 +358,7 @@ impl FastBookLevel {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Full order book state
|
/// Full order book state
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
pub struct OrderBook {
|
pub struct OrderBook {
|
||||||
/// Token ID
|
/// Token ID
|
||||||
pub token_id: String,
|
pub token_id: String,
|
||||||
@@ -368,8 +368,47 @@ pub struct OrderBook {
|
|||||||
pub bids: Vec<BookLevel>,
|
pub bids: Vec<BookLevel>,
|
||||||
/// Ask orders
|
/// Ask orders
|
||||||
pub asks: Vec<BookLevel>,
|
pub asks: Vec<BookLevel>,
|
||||||
/// Sequence number
|
/// Legacy/incremental delta sequence number.
|
||||||
|
///
|
||||||
|
/// This is kept for compatibility. For snapshots produced by this client,
|
||||||
|
/// it is equivalent to [`Self::last_delta_sequence`]. WebSocket snapshot
|
||||||
|
/// timestamps are exposed separately in [`Self::last_snapshot_timestamp_ms`].
|
||||||
pub sequence: u64,
|
pub sequence: u64,
|
||||||
|
/// Last accepted legacy/incremental delta sequence number.
|
||||||
|
pub last_delta_sequence: u64,
|
||||||
|
/// Last accepted full-book snapshot timestamp in milliseconds.
|
||||||
|
pub last_snapshot_timestamp_ms: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<'de> Deserialize<'de> for OrderBook {
|
||||||
|
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
|
||||||
|
where
|
||||||
|
D: serde::Deserializer<'de>,
|
||||||
|
{
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
struct WireOrderBook {
|
||||||
|
token_id: String,
|
||||||
|
timestamp: DateTime<Utc>,
|
||||||
|
bids: Vec<BookLevel>,
|
||||||
|
asks: Vec<BookLevel>,
|
||||||
|
sequence: u64,
|
||||||
|
#[serde(default)]
|
||||||
|
last_delta_sequence: Option<u64>,
|
||||||
|
#[serde(default)]
|
||||||
|
last_snapshot_timestamp_ms: Option<u64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
let wire = WireOrderBook::deserialize(deserializer)?;
|
||||||
|
Ok(Self {
|
||||||
|
token_id: wire.token_id,
|
||||||
|
timestamp: wire.timestamp,
|
||||||
|
bids: wire.bids,
|
||||||
|
asks: wire.asks,
|
||||||
|
sequence: wire.sequence,
|
||||||
|
last_delta_sequence: wire.last_delta_sequence.unwrap_or(wire.sequence),
|
||||||
|
last_snapshot_timestamp_ms: wire.last_snapshot_timestamp_ms.unwrap_or_default(),
|
||||||
|
})
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Order book delta for streaming updates - EXTERNAL API VERSION
|
/// Order book delta for streaming updates - EXTERNAL API VERSION
|
||||||
@@ -937,16 +976,26 @@ pub enum StreamMessage {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Orderbook update message (full snapshot or delta).
|
/// Orderbook update message (full snapshot or delta).
|
||||||
|
///
|
||||||
|
/// WebSocket `book` messages expose a millisecond timestamp and optional book hash, but no
|
||||||
|
/// monotonic server sequence/version. Same-timestamp messages with different hashes are therefore
|
||||||
|
/// ordered by websocket arrival order by the book applier; the hash is a duplicate/state
|
||||||
|
/// discriminator, not a logical ordering key.
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||||
pub struct BookUpdate {
|
pub struct BookUpdate {
|
||||||
pub asset_id: String,
|
pub asset_id: String,
|
||||||
pub market: String,
|
pub market: String,
|
||||||
|
/// Exchange-provided snapshot timestamp in milliseconds.
|
||||||
#[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
|
#[serde(deserialize_with = "crate::decode::deserializers::number_from_string")]
|
||||||
pub timestamp: u64,
|
pub timestamp: u64,
|
||||||
#[serde(deserialize_with = "crate::decode::deserializers::vec_from_null")]
|
#[serde(deserialize_with = "crate::decode::deserializers::vec_from_null")]
|
||||||
pub bids: Vec<OrderSummary>,
|
pub bids: Vec<OrderSummary>,
|
||||||
#[serde(deserialize_with = "crate::decode::deserializers::vec_from_null")]
|
#[serde(deserialize_with = "crate::decode::deserializers::vec_from_null")]
|
||||||
pub asks: Vec<OrderSummary>,
|
pub asks: Vec<OrderSummary>,
|
||||||
|
/// Exchange-provided snapshot hash.
|
||||||
|
///
|
||||||
|
/// Used to suppress exact duplicate same-timestamp snapshots and to allow distinct
|
||||||
|
/// same-timestamp states. It does not encode ordering.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub hash: Option<String>,
|
pub hash: Option<String>,
|
||||||
}
|
}
|
||||||
@@ -1922,4 +1971,20 @@ mod tests {
|
|||||||
Decimal::from_str("0.00005").unwrap()
|
Decimal::from_str("0.00005").unwrap()
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn order_book_snapshot_clocks_default_for_legacy_payloads() {
|
||||||
|
let json = r#"{
|
||||||
|
"token_id":"test_token",
|
||||||
|
"timestamp":"2026-06-23T00:00:00Z",
|
||||||
|
"bids":[],
|
||||||
|
"asks":[],
|
||||||
|
"sequence":42
|
||||||
|
}"#;
|
||||||
|
|
||||||
|
let book: OrderBook = serde_json::from_str(json).unwrap();
|
||||||
|
assert_eq!(book.sequence, 42);
|
||||||
|
assert_eq!(book.last_delta_sequence, 42);
|
||||||
|
assert_eq!(book.last_snapshot_timestamp_ms, 0);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -385,6 +385,8 @@ mod tests {
|
|||||||
|
|
||||||
let snapshot = books.get_book("test_asset_id").unwrap();
|
let snapshot = books.get_book("test_asset_id").unwrap();
|
||||||
assert_eq!(snapshot.sequence, 0);
|
assert_eq!(snapshot.sequence, 0);
|
||||||
|
assert_eq!(snapshot.last_delta_sequence, 0);
|
||||||
|
assert_eq!(snapshot.last_snapshot_timestamp_ms, 1000);
|
||||||
assert_eq!(snapshot.timestamp.timestamp_millis(), 1000);
|
assert_eq!(snapshot.timestamp.timestamp_millis(), 1000);
|
||||||
assert_eq!(snapshot.bids.len(), 1);
|
assert_eq!(snapshot.bids.len(), 1);
|
||||||
assert_eq!(snapshot.asks.len(), 1);
|
assert_eq!(snapshot.asks.len(), 1);
|
||||||
@@ -423,6 +425,8 @@ mod tests {
|
|||||||
|
|
||||||
let snapshot = books.get_book("test_asset_id").unwrap();
|
let snapshot = books.get_book("test_asset_id").unwrap();
|
||||||
assert_eq!(snapshot.sequence, 0);
|
assert_eq!(snapshot.sequence, 0);
|
||||||
|
assert_eq!(snapshot.last_delta_sequence, 0);
|
||||||
|
assert_eq!(snapshot.last_snapshot_timestamp_ms, 1000);
|
||||||
assert_eq!(snapshot.timestamp.timestamp_millis(), 1000);
|
assert_eq!(snapshot.timestamp.timestamp_millis(), 1000);
|
||||||
assert_eq!(snapshot.bids.len(), 1);
|
assert_eq!(snapshot.bids.len(), 1);
|
||||||
assert_eq!(snapshot.asks.len(), 1);
|
assert_eq!(snapshot.asks.len(), 1);
|
||||||
@@ -450,6 +454,8 @@ mod tests {
|
|||||||
|
|
||||||
let snapshot = books.get_book("test_asset_id").unwrap();
|
let snapshot = books.get_book("test_asset_id").unwrap();
|
||||||
assert_eq!(snapshot.sequence, 0);
|
assert_eq!(snapshot.sequence, 0);
|
||||||
|
assert_eq!(snapshot.last_delta_sequence, 0);
|
||||||
|
assert_eq!(snapshot.last_snapshot_timestamp_ms, 1000);
|
||||||
assert_eq!(snapshot.timestamp.timestamp_millis(), 1000);
|
assert_eq!(snapshot.timestamp.timestamp_millis(), 1000);
|
||||||
assert_eq!(snapshot.bids[0].price, dec!(0.51));
|
assert_eq!(snapshot.bids[0].price, dec!(0.51));
|
||||||
assert_eq!(snapshot.bids[0].size, dec!(11));
|
assert_eq!(snapshot.bids[0].size, dec!(11));
|
||||||
|
|||||||
Reference in New Issue
Block a user