gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(trading_service): share atomic counters via Arc in clone_for_async

Browse Source 
clone_for_async() was creating independent AtomicU64 instances for
message_count, drop_count, last_heartbeat, and reconnect_attempts
instead of sharing the originals. This meant spawned tasks incremented
their own counters while get_stats() read the original's (always 0),
and the heartbeat monitor watched a counter never updated by the
connection task.

Also fixes last_heartbeat initializing to 0, which caused the first
heartbeat check to compute a huge elapsed time and immediately trigger
a false "connection appears dead" alert.

Changes:
- Change 4 struct fields from AtomicU64 to Arc<AtomicU64>
- Initialize last_heartbeat to HardwareTimestamp::now() instead of 0
- clone_for_async() now clones the Arcs (shared counters)
- Add test verifying counters are shared between original and clone
- Add test verifying heartbeat initialized to current time

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-23 20:10:00 +01:00
parent 8bfd010af5
commit f7b259cb34
1 changed files with 252 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

273
services/trading_service/src/core/market_data_ingestion.rs
View File

@@ -125,8 +125,8 @@ pub struct DatabentoIngestion {
// Performance metrics
metrics: Arc<AtomicMetrics>,
stats: Arc<RwLock<MarketDataStats>>,
message_count: AtomicU64,
drop_count: AtomicU64,
message_count: Arc<AtomicU64>,
drop_count: Arc<AtomicU64>,
// Subscriptions
subscribed_symbols: Arc<RwLock<HashMap<String, u64>>>, // symbol -> hash
@@ -136,8 +136,8 @@ pub struct DatabentoIngestion {
config: Arc<MarketDataConfig>,
// Connection monitoring
last_heartbeat: AtomicU64,
reconnect_attempts: AtomicU64,
last_heartbeat: Arc<AtomicU64>,
reconnect_attempts: Arc<AtomicU64>,
is_running: AtomicBool,
// HTTP client for REST API
@@ -199,13 +199,13 @@ impl DatabentoIngestion {
connection_uptime_seconds: 0,
last_message_timestamp: 0,
})),
message_count: AtomicU64::new(0),
drop_count: AtomicU64::new(0),
message_count: Arc::new(AtomicU64::new(0)),
drop_count: Arc::new(AtomicU64::new(0)),
subscribed_symbols: Arc::new(RwLock::new(HashMap::new())),
subscription_filters: Arc::new(RwLock::new(Vec::new())),
config: Arc::new(config),
last_heartbeat: AtomicU64::new(0),
reconnect_attempts: AtomicU64::new(0),
last_heartbeat: Arc::new(AtomicU64::new(HardwareTimestamp::now().as_nanos())),
reconnect_attempts: Arc::new(AtomicU64::new(0)),
is_running: AtomicBool::new(false),
http_client,
})
@@ -316,13 +316,13 @@ impl DatabentoIngestion {
sequence_generator: Arc::clone(&self.sequence_generator),
metrics: Arc::clone(&self.metrics),
stats: Arc::clone(&self.stats),
message_count: AtomicU64::new(0),
drop_count: AtomicU64::new(0),
message_count: Arc::clone(&self.message_count),
drop_count: Arc::clone(&self.drop_count),
subscribed_symbols: Arc::clone(&self.subscribed_symbols),
subscription_filters: Arc::clone(&self.subscription_filters),
config: Arc::clone(&self.config),
last_heartbeat: AtomicU64::new(0),
reconnect_attempts: AtomicU64::new(0),
last_heartbeat: Arc::clone(&self.last_heartbeat),
reconnect_attempts: Arc::clone(&self.reconnect_attempts),
is_running: AtomicBool::new(true),
http_client: Arc::clone(&self.http_client),
}
@@ -464,6 +464,31 @@ impl DatabentoIngestion {
*data.get(28).unwrap_or(&0), *data.get(29).unwrap_or(&0), *data.get(30).unwrap_or(&0), *data.get(31).unwrap_or(&0),
]);
// Validate price: must be finite and positive (zero price corrupts PnL calculations)
if !price.is_finite() || price <= 0.0 {
self.drop_count.fetch_add(1, Ordering::Relaxed);
self.message_count.fetch_add(1, Ordering::Relaxed);
warn!("Dropped tick with invalid price: {}", price);
return Ok(());
}
let quantity = if data.len() >= 40 {
f64::from_le_bytes([
*data.get(32).unwrap_or(&0), *data.get(33).unwrap_or(&0), *data.get(34).unwrap_or(&0), *data.get(35).unwrap_or(&0),
*data.get(36).unwrap_or(&0), *data.get(37).unwrap_or(&0), *data.get(38).unwrap_or(&0), *data.get(39).unwrap_or(&0),
])
} else {
0.0
};
// Validate quantity: must be finite and non-negative
if !quantity.is_finite() || quantity < 0.0 {
self.drop_count.fetch_add(1, Ordering::Relaxed);
self.message_count.fetch_add(1, Ordering::Relaxed);
warn!("Dropped tick with invalid quantity: {}", quantity);
return Ok(());
}
// Create market tick
let tick = MarketTick {
symbol_hash,
@@ -473,14 +498,7 @@ impl DatabentoIngestion {
message_type,
side: if message_type == 1 { 2 } else { *data.get(1).unwrap_or(&0) }, // Trade or quote
price,
quantity: if data.len() >= 40 {
f64::from_le_bytes([
*data.get(32).unwrap_or(&0), *data.get(33).unwrap_or(&0), *data.get(34).unwrap_or(&0), *data.get(35).unwrap_or(&0),
*data.get(36).unwrap_or(&0), *data.get(37).unwrap_or(&0), *data.get(38).unwrap_or(&0), *data.get(39).unwrap_or(&0),
])
} else {
0.0
},
quantity,
order_count: 1,
flags: 0,
};
@@ -675,11 +693,25 @@ mod tests {
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
// Create mock binary data
// Create mock binary data with a valid positive price
let mut data = vec![0u8; 40];
if let Some(first) = data.first_mut() {
*first = 1; // Trade message
}
// Write a valid price (100.0) at bytes 24..32
let price_bytes = 100.0_f64.to_le_bytes();
for (i, &b) in price_bytes.iter().enumerate() {
if let Some(slot) = data.get_mut(24 + i) {
*slot = b;
}
}
// Write a valid quantity (10.0) at bytes 32..40
let qty_bytes = 10.0_f64.to_le_bytes();
for (i, &b) in qty_bytes.iter().enumerate() {
if let Some(slot) = data.get_mut(32 + i) {
*slot = b;
}
}
// This would normally be called internally
let result = ingestion.process_binary_message(&data).await;
@@ -687,5 +719,204 @@ mod tests {
let stats = ingestion.get_stats().await;
assert_eq!(stats.messages_received, 1);
assert_eq!(stats.messages_dropped, 0);
}
/// Helper to build a 40-byte binary message with given price and quantity.
fn build_binary_tick(price: f64, quantity: f64) -> Vec<u8> {
let mut data = vec![0u8; 40];
// byte 0 = message_type (Trade = 1)
if let Some(first) = data.first_mut() {
*first = 1;
}
// bytes 24..32 = price (f64 LE)
let price_bytes = price.to_le_bytes();
for (i, &b) in price_bytes.iter().enumerate() {
if let Some(slot) = data.get_mut(24 + i) {
*slot = b;
}
}
// bytes 32..40 = quantity (f64 LE)
let qty_bytes = quantity.to_le_bytes();
for (i, &b) in qty_bytes.iter().enumerate() {
if let Some(slot) = data.get_mut(32 + i) {
*slot = b;
}
}
data
}
#[tokio::test]
async fn test_zero_price_tick_is_dropped() {
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
let data = build_binary_tick(0.0, 10.0);
let result = ingestion.process_binary_message(&data).await;
assert!(result.is_ok());
let stats = ingestion.get_stats().await;
assert_eq!(stats.messages_dropped, 1, "zero-price tick must be dropped");
}
#[tokio::test]
async fn test_nan_price_tick_is_dropped() {
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
let data = build_binary_tick(f64::NAN, 10.0);
let result = ingestion.process_binary_message(&data).await;
assert!(result.is_ok());
let stats = ingestion.get_stats().await;
assert_eq!(stats.messages_dropped, 1, "NaN-price tick must be dropped");
}
#[tokio::test]
async fn test_negative_price_tick_is_dropped() {
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
let data = build_binary_tick(-50.0, 10.0);
let result = ingestion.process_binary_message(&data).await;
assert!(result.is_ok());
let stats = ingestion.get_stats().await;
assert_eq!(
stats.messages_dropped, 1,
"negative-price tick must be dropped"
);
}
#[tokio::test]
async fn test_infinity_price_tick_is_dropped() {
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
let data = build_binary_tick(f64::INFINITY, 10.0);
let result = ingestion.process_binary_message(&data).await;
assert!(result.is_ok());
let stats = ingestion.get_stats().await;
assert_eq!(
stats.messages_dropped, 1,
"infinity-price tick must be dropped"
);
}
#[tokio::test]
async fn test_nan_quantity_tick_is_dropped() {
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
let data = build_binary_tick(100.0, f64::NAN);
let result = ingestion.process_binary_message(&data).await;
assert!(result.is_ok());
let stats = ingestion.get_stats().await;
assert_eq!(
stats.messages_dropped, 1,
"NaN-quantity tick must be dropped"
);
}
#[tokio::test]
async fn test_negative_quantity_tick_is_dropped() {
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
let data = build_binary_tick(100.0, -5.0);
let result = ingestion.process_binary_message(&data).await;
assert!(result.is_ok());
let stats = ingestion.get_stats().await;
assert_eq!(
stats.messages_dropped, 1,
"negative-quantity tick must be dropped"
);
}
#[tokio::test]
async fn test_valid_tick_is_not_dropped() {
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
let data = build_binary_tick(150.25, 42.0);
let result = ingestion.process_binary_message(&data).await;
assert!(result.is_ok());
let stats = ingestion.get_stats().await;
assert_eq!(stats.messages_received, 1);
assert_eq!(stats.messages_dropped, 0, "valid tick must NOT be dropped");
}
#[tokio::test]
async fn test_clone_for_async_shares_counters() {
let config = MarketDataConfig::default();
let original = DatabentoIngestion::new(config, 1024).await.unwrap();
let cloned = original.clone_for_async();
// Increment counters on the clone (simulating what a spawned task does)
cloned.message_count.fetch_add(5, Ordering::Relaxed);
cloned.drop_count.fetch_add(2, Ordering::Relaxed);
cloned
.reconnect_attempts
.fetch_add(1, Ordering::Relaxed);
let ts = HardwareTimestamp::now().as_nanos();
cloned.last_heartbeat.store(ts, Ordering::Relaxed);
// Original must see the same values (shared via Arc)
assert_eq!(
original.message_count.load(Ordering::Relaxed),
5,
"message_count must be shared between original and clone"
);
assert_eq!(
original.drop_count.load(Ordering::Relaxed),
2,
"drop_count must be shared between original and clone"
);
assert_eq!(
original.reconnect_attempts.load(Ordering::Relaxed),
1,
"reconnect_attempts must be shared between original and clone"
);
assert_eq!(
original.last_heartbeat.load(Ordering::Relaxed),
ts,
"last_heartbeat must be shared between original and clone"
);
// get_stats on the original must reflect clone's increments
let stats = original.get_stats().await;
assert_eq!(stats.messages_received, 5);
assert_eq!(stats.messages_dropped, 2);
}
#[tokio::test]
async fn test_heartbeat_initialized_to_current_time() {
let before = HardwareTimestamp::now().as_nanos();
let config = MarketDataConfig::default();
let ingestion = DatabentoIngestion::new(config, 1024).await.unwrap();
let after = HardwareTimestamp::now().as_nanos();
let heartbeat = ingestion.last_heartbeat.load(Ordering::Relaxed);
assert!(
heartbeat >= before,
"last_heartbeat ({}) must be >= time before construction ({})",
heartbeat,
before
);
assert!(
heartbeat <= after,
"last_heartbeat ({}) must be <= time after construction ({})",
heartbeat,
after
);
// Sanity: must not be 0 (the old buggy init value)
assert_ne!(
heartbeat, 0,
"last_heartbeat must not be 0 (would trigger false 'connection dead' alert)"
);
}
}