use std::alloc::{GlobalAlloc, Layout, System};
use std::cell::Cell;
use std::collections::hash_map::DefaultHasher;
use std::fmt::Write as _;
use std::hash::{Hash, Hasher};
use std::str::FromStr;
use chrono::Utc;
use polyfill_rs::{
book::OrderBookManager, OrderBookImpl, Side, WebSocketStream, WsBookUpdateProcessor,
};
use rust_decimal::Decimal;
thread_local! {
static HEAP_OPERATIONS: Cell<usize> = const { Cell::new(0) };
}
struct CountingAllocator;
unsafe impl GlobalAlloc for CountingAllocator {
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
HEAP_OPERATIONS.with(|count| count.set(count.get() + 1));
System.alloc(layout)
}
unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 {
HEAP_OPERATIONS.with(|count| count.set(count.get() + 1));
System.alloc_zeroed(layout)
}
unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
HEAP_OPERATIONS.with(|count| count.set(count.get() + 1));
System.realloc(ptr, layout, new_size)
}
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
HEAP_OPERATIONS.with(|count| count.set(count.get() + 1));
System.dealloc(ptr, layout)
}
}
#[global_allocator]
static GLOBAL: CountingAllocator = CountingAllocator;
fn heap_operation_count() -> usize {
HEAP_OPERATIONS.with(|count| count.get())
}
struct NoHeapTrafficGuard {
before: usize,
}
impl NoHeapTrafficGuard {
fn new() -> Self {
Self {
before: heap_operation_count(),
}
}
fn assert_no_heap_traffic(self) {
let after = heap_operation_count();
assert_eq!(
after,
self.before,
"expected no heap traffic, but saw {} allocator operation(s)",
after - self.before
);
}
}
#[test]
fn allocator_counter_tracks_deallocations() {
let vec = Vec::<u8>::with_capacity(1024);
let before = heap_operation_count();
drop(vec);
let after = heap_operation_count();
assert!(
after > before,
"expected dropping an allocated Vec to count as heap traffic"
);
}
fn token_id_hash(token_id: &str) -> u64 {
let mut hasher = DefaultHasher::new();
token_id.hash(&mut hasher);
hasher.finish()
}
fn mk_delta(
token_id_hash: u64,
side: Side,
price_ticks: polyfill_rs::types::Price,
size_units: polyfill_rs::types::Qty,
sequence: u64,
) -> polyfill_rs::types::FastOrderDelta {
polyfill_rs::types::FastOrderDelta {
token_id_hash,
timestamp: chrono::DateTime::<Utc>::from_timestamp(0, 0).unwrap(),
side,
price: price_ticks,
size: size_units,
sequence,
}
}
fn seed_book_levels(
manager: &OrderBookManager,
asset_id: &str,
bid_ticks: &[i64],
ask_ticks: &[i64],
) {
let mut sequence = 1u64;
for &price_ticks in bid_ticks {
manager
.apply_delta(polyfill_rs::types::OrderDelta {
token_id: asset_id.to_string(),
timestamp: chrono::Utc::now(),
side: Side::BUY,
price: Decimal::new(price_ticks, 4),
size: Decimal::from_str("100.0").unwrap(),
sequence,
})
.unwrap();
sequence += 1;
}
for &price_ticks in ask_ticks {
manager
.apply_delta(polyfill_rs::types::OrderDelta {
token_id: asset_id.to_string(),
timestamp: chrono::Utc::now(),
side: Side::SELL,
price: Decimal::new(price_ticks, 4),
size: Decimal::from_str("100.0").unwrap(),
sequence,
})
.unwrap();
sequence += 1;
}
}
fn ws_book_message(
asset_id: &str,
timestamp: u64,
bid_ticks: &[i64],
ask_ticks: &[i64],
) -> Vec<u8> {
let mut json = String::with_capacity(160 + (bid_ticks.len() + ask_ticks.len()) * 40);
write!(
&mut json,
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":{timestamp},\"bids\":["
)
.unwrap();
for (idx, price_ticks) in bid_ticks.iter().enumerate() {
if idx > 0 {
json.push(',');
}
write!(
&mut json,
"{{\"price\":\"{}\",\"size\":\"100.0000\"}}",
Decimal::new(*price_ticks, 4)
)
.unwrap();
}
json.push_str("],\"asks\":[");
for (idx, price_ticks) in ask_ticks.iter().enumerate() {
if idx > 0 {
json.push(',');
}
write!(
&mut json,
"{{\"price\":\"{}\",\"size\":\"100.0000\"}}",
Decimal::new(*price_ticks, 4)
)
.unwrap();
}
json.push_str("]}");
json.into_bytes()
}
fn ws_book_message_with_hash(
asset_id: &str,
timestamp: u64,
hash: &str,
bid_ticks: &[i64],
ask_ticks: &[i64],
) -> Vec<u8> {
let mut json =
String::with_capacity(190 + hash.len() + (bid_ticks.len() + ask_ticks.len()) * 40);
write!(
&mut json,
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":{timestamp},\"hash\":\"{hash}\",\"bids\":["
)
.unwrap();
for (idx, price_ticks) in bid_ticks.iter().enumerate() {
if idx > 0 {
json.push(',');
}
write!(
&mut json,
"{{\"price\":\"{}\",\"size\":\"100.0000\"}}",
Decimal::new(*price_ticks, 4)
)
.unwrap();
}
json.push_str("],\"asks\":[");
for (idx, price_ticks) in ask_ticks.iter().enumerate() {
if idx > 0 {
json.push(',');
}
write!(
&mut json,
"{{\"price\":\"{}\",\"size\":\"100.0000\"}}",
Decimal::new(*price_ticks, 4)
)
.unwrap();
}
json.push_str("]}");
json.into_bytes()
}
fn contiguous_ticks(start: i64, len: usize, step: i64) -> Vec<i64> {
(0..len).map(|idx| start + (idx as i64 * step)).collect()
}
#[test]
fn no_alloc_mid_and_spread_fast() {
let token_id = "test_token";
let token_hash = token_id_hash(token_id);
let mut book = OrderBookImpl::new(token_id.to_string(), 100);
book.apply_delta_fast(mk_delta(token_hash, Side::BUY, 7500, 1_000_000, 1))
.unwrap();
book.apply_delta_fast(mk_delta(token_hash, Side::SELL, 7600, 1_000_000, 2))
.unwrap();
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
assert!(book.best_bid_fast().is_some());
assert!(book.best_ask_fast().is_some());
assert!(book.spread_fast().is_some());
assert!(book.mid_price_fast().is_some());
guard.assert_no_heap_traffic();
}
#[test]
fn no_alloc_book_analysis_fast_paths() {
let token_id = "test_token";
let token_hash = token_id_hash(token_id);
let mut book = OrderBookImpl::new(token_id.to_string(), 100);
book.apply_delta_fast(mk_delta(token_hash, Side::BUY, 7500, 1_000_000, 1))
.unwrap();
book.apply_delta_fast(mk_delta(token_hash, Side::BUY, 7400, 500_000, 2))
.unwrap();
book.apply_delta_fast(mk_delta(token_hash, Side::SELL, 7600, 800_000, 3))
.unwrap();
book.apply_delta_fast(mk_delta(token_hash, Side::SELL, 7700, 1_200_000, 4))
.unwrap();
let impact_size = Decimal::from_str("150.0").unwrap();
let min_price = Decimal::from_str("0.74").unwrap();
let max_price = Decimal::from_str("0.77").unwrap();
let min_average_price = Decimal::from_str("0.76").unwrap();
let expected_buy_liquidity = Decimal::from_str("200.0").unwrap();
let expected_sell_liquidity = Decimal::from_str("150.0").unwrap();
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
let impact = book
.calculate_market_impact(Side::BUY, impact_size)
.unwrap();
assert!(impact.average_price > min_average_price);
assert_eq!(
book.liquidity_in_range(min_price, max_price, Side::BUY),
expected_buy_liquidity
);
assert_eq!(
book.liquidity_in_range(min_price, max_price, Side::SELL),
expected_sell_liquidity
);
assert!(book.is_valid());
guard.assert_no_heap_traffic();
}
#[test]
fn no_alloc_apply_delta_fast_existing_level_update() {
let token_id = "test_token";
let token_hash = token_id_hash(token_id);
let mut book = OrderBookImpl::new(token_id.to_string(), 100);
book.apply_delta_fast(mk_delta(token_hash, Side::BUY, 7500, 1_000_000, 1))
.unwrap();
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
book.apply_delta_fast(mk_delta(token_hash, Side::BUY, 7500, 2_000_000, 2))
.unwrap();
guard.assert_no_heap_traffic();
}
#[test]
fn no_alloc_apply_book_update_existing_levels() {
let asset_id = "test_asset_id";
let token_hash = token_id_hash(asset_id);
let mut book = OrderBookImpl::new(asset_id.to_string(), 100);
book.apply_delta_fast(mk_delta(token_hash, Side::BUY, 7500, 1_000_000, 1))
.unwrap();
book.apply_delta_fast(mk_delta(token_hash, Side::SELL, 7600, 1_000_000, 2))
.unwrap();
let update = polyfill_rs::types::BookUpdate {
asset_id: asset_id.to_string(),
market: "0xabc".to_string(),
timestamp: 10,
bids: vec![polyfill_rs::types::OrderSummary {
price: Decimal::from_str("0.75").unwrap(),
size: Decimal::from_str("200.0").unwrap(),
}],
asks: vec![polyfill_rs::types::OrderSummary {
price: Decimal::from_str("0.76").unwrap(),
size: Decimal::from_str("50.0").unwrap(),
}],
hash: None,
};
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
book.apply_book_update(&update).unwrap();
guard.assert_no_heap_traffic();
}
#[test]
fn no_alloc_book_manager_apply_book_update_existing_levels() {
let asset_id = "test_asset_id";
let manager = OrderBookManager::new(100);
manager.get_or_create_book(asset_id).unwrap();
manager
.apply_delta(polyfill_rs::types::OrderDelta {
token_id: asset_id.to_string(),
timestamp: chrono::Utc::now(),
side: Side::BUY,
price: Decimal::from_str("0.75").unwrap(),
size: Decimal::from_str("100.0").unwrap(),
sequence: 1,
})
.unwrap();
manager
.apply_delta(polyfill_rs::types::OrderDelta {
token_id: asset_id.to_string(),
timestamp: chrono::Utc::now(),
side: Side::SELL,
price: Decimal::from_str("0.76").unwrap(),
size: Decimal::from_str("100.0").unwrap(),
sequence: 2,
})
.unwrap();
let update = polyfill_rs::types::BookUpdate {
asset_id: asset_id.to_string(),
market: "0xabc".to_string(),
timestamp: 10,
bids: vec![polyfill_rs::types::OrderSummary {
price: Decimal::from_str("0.75").unwrap(),
size: Decimal::from_str("200.0").unwrap(),
}],
asks: vec![polyfill_rs::types::OrderSummary {
price: Decimal::from_str("0.76").unwrap(),
size: Decimal::from_str("50.0").unwrap(),
}],
hash: None,
};
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
manager.apply_book_update(&update).unwrap();
guard.assert_no_heap_traffic();
}
#[test]
fn no_alloc_ws_book_update_processor_apply_existing_levels() {
let asset_id = "test_asset_id";
let manager = OrderBookManager::new(100);
manager.get_or_create_book(asset_id).unwrap();
manager
.apply_delta(polyfill_rs::types::OrderDelta {
token_id: asset_id.to_string(),
timestamp: chrono::Utc::now(),
side: Side::BUY,
price: Decimal::from_str("0.75").unwrap(),
size: Decimal::from_str("100.0").unwrap(),
sequence: 1,
})
.unwrap();
manager
.apply_delta(polyfill_rs::types::OrderDelta {
token_id: asset_id.to_string(),
timestamp: chrono::Utc::now(),
side: Side::SELL,
price: Decimal::from_str("0.76").unwrap(),
size: Decimal::from_str("100.0").unwrap(),
sequence: 2,
})
.unwrap();
let mut processor = WsBookUpdateProcessor::new(1024);
let mut warmup_msg = format!(
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":10,\"bids\":[{{\"price\":\"0.75\",\"size\":\"200.0\"}}],\"asks\":[{{\"price\":\"0.76\",\"size\":\"50.0\"}}]}}"
)
.into_bytes();
processor
.process_bytes(warmup_msg.as_mut_slice(), &manager)
.unwrap();
let mut msg = format!(
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":11,\"bids\":[{{\"price\":\"0.75\",\"size\":\"150.0\"}}],\"asks\":[{{\"price\":\"0.76\",\"size\":\"75.0\"}}]}}"
)
.into_bytes();
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
processor
.process_bytes(msg.as_mut_slice(), &manager)
.unwrap();
guard.assert_no_heap_traffic();
}
#[test]
fn no_alloc_ws_book_update_processor_same_timestamp_hash_change() {
let asset_id = "test_asset_id";
let manager = OrderBookManager::new(100);
manager.get_or_create_book(asset_id).unwrap();
seed_book_levels(&manager, asset_id, &[7500], &[7600]);
let mut processor = WsBookUpdateProcessor::new(4096);
let mut warmup_msg = ws_book_message_with_hash(asset_id, 10, "hash_a", &[7500], &[7600]);
processor
.process_bytes(warmup_msg.as_mut_slice(), &manager)
.unwrap();
let mut msg = ws_book_message_with_hash(asset_id, 10, "hash_b", &[7500], &[7600]);
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
processor
.process_bytes(msg.as_mut_slice(), &manager)
.unwrap();
guard.assert_no_heap_traffic();
}
#[test]
fn no_alloc_ws_book_update_processor_one_new_level_with_reserved_capacity() {
let asset_id = "test_asset_id";
let manager = OrderBookManager::new(100);
manager.get_or_create_book(asset_id).unwrap();
seed_book_levels(&manager, asset_id, &[7500], &[7600]);
let mut processor = WsBookUpdateProcessor::new(4096);
let mut warmup_msg = ws_book_message(asset_id, 10, &[7500], &[7600]);
processor
.process_bytes(warmup_msg.as_mut_slice(), &manager)
.unwrap();
let mut parser_capacity_warmup = ws_book_message(asset_id, 10, &[7500, 7400], &[7600]);
processor
.process_bytes(parser_capacity_warmup.as_mut_slice(), &manager)
.unwrap();
let mut msg = ws_book_message(asset_id, 11, &[7500, 7400], &[7600]);
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
processor
.process_bytes(msg.as_mut_slice(), &manager)
.unwrap();
guard.assert_no_heap_traffic();
}
#[test]
fn no_alloc_ws_book_update_processor_one_removed_level() {
let asset_id = "test_asset_id";
let manager = OrderBookManager::new(100);
manager.get_or_create_book(asset_id).unwrap();
seed_book_levels(&manager, asset_id, &[7500, 7400], &[7600]);
let mut processor = WsBookUpdateProcessor::new(4096);
let mut warmup_msg = ws_book_message(asset_id, 10, &[7500, 7400], &[7600]);
processor
.process_bytes(warmup_msg.as_mut_slice(), &manager)
.unwrap();
let mut msg = ws_book_message(asset_id, 11, &[7500], &[7600]);
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
processor
.process_bytes(msg.as_mut_slice(), &manager)
.unwrap();
guard.assert_no_heap_traffic();
}
#[test]
fn ws_book_update_processor_full_churn_64_levels_touches_allocator() {
let asset_id = "test_asset_id";
let levels_per_side = 64;
let manager = OrderBookManager::new(levels_per_side);
manager.get_or_create_book(asset_id).unwrap();
let initial_bids = contiguous_ticks(7500, levels_per_side, -1);
let initial_asks = contiguous_ticks(7600, levels_per_side, 1);
seed_book_levels(&manager, asset_id, &initial_bids, &initial_asks);
let mut warmup_msg = ws_book_message(asset_id, 1000, &initial_bids, &initial_asks);
let mut processor = WsBookUpdateProcessor::new(warmup_msg.len());
processor
.process_bytes(warmup_msg.as_mut_slice(), &manager)
.unwrap();
let churn_bids = contiguous_ticks(7300, levels_per_side, -1);
let churn_asks = contiguous_ticks(7800, levels_per_side, 1);
let mut msg = ws_book_message(asset_id, 1001, &churn_bids, &churn_asks);
let _ = heap_operation_count();
let before = heap_operation_count();
processor
.process_bytes(msg.as_mut_slice(), &manager)
.unwrap();
let after = heap_operation_count();
assert!(
after > before,
"expected 64-level full churn to touch the allocator while old and new levels coexist"
);
}
#[test]
fn no_alloc_websocket_book_applier_apply_bytes_message_existing_levels() {
let asset_id = "test_asset_id";
let manager = OrderBookManager::new(100);
manager.get_or_create_book(asset_id).unwrap();
manager
.apply_delta(polyfill_rs::types::OrderDelta {
token_id: asset_id.to_string(),
timestamp: chrono::Utc::now(),
side: Side::BUY,
price: Decimal::from_str("0.75").unwrap(),
size: Decimal::from_str("100.0").unwrap(),
sequence: 1,
})
.unwrap();
manager
.apply_delta(polyfill_rs::types::OrderDelta {
token_id: asset_id.to_string(),
timestamp: chrono::Utc::now(),
side: Side::SELL,
price: Decimal::from_str("0.76").unwrap(),
size: Decimal::from_str("100.0").unwrap(),
sequence: 2,
})
.unwrap();
let processor = WsBookUpdateProcessor::new(1024);
let stream = WebSocketStream::new("wss://example.com/ws");
let mut applier = stream.into_book_applier(&manager, processor);
let mut warmup_msg = format!(
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":10,\"bids\":[{{\"price\":\"0.75\",\"size\":\"200.0\"}}],\"asks\":[{{\"price\":\"0.76\",\"size\":\"50.0\"}}]}}"
)
.into_bytes();
applier
.apply_bytes_message(warmup_msg.as_mut_slice())
.unwrap();
let mut msg = format!(
"{{\"event_type\":\"book\",\"asset_id\":\"{asset_id}\",\"market\":\"0xabc\",\"timestamp\":11,\"bids\":[{{\"price\":\"0.75\",\"size\":\"150.0\"}}],\"asks\":[{{\"price\":\"0.76\",\"size\":\"75.0\"}}]}}"
)
.into_bytes();
let _ = heap_operation_count();
let guard = NoHeapTrafficGuard::new();
applier.apply_bytes_message(msg.as_mut_slice()).unwrap();
guard.assert_no_heap_traffic();
}