eae3c31e53
Patterns applied: - Pattern 2: Float comparison (2x: utils.rs, var_edge_cases_tests.rs) - Pattern 7: Date/time construction (2x: production_streaming.rs, streaming.rs) - Pattern 1: Duration/time ops (2x: rate limiter, semaphore) - Pattern 4: Optional field access (1x: position_tracker.rs) Changes: - data/src/utils.rs: Float sort with NaN handling - data/src/providers/benzinga/production_streaming.rs: Rate limiter + semaphore + date/time - data/src/providers/benzinga/streaming.rs: Date/time construction - risk/src/position_tracker.rs: Emergency fallback counter - risk/tests/var_edge_cases_tests.rs: Test helper float sort Test impact: 0 failures (182/182 passing) Compilation: Clean (0 errors, 0 warnings) Time: 25 min (44% under budget)
2085 lines
66 KiB
Rust
2085 lines
66 KiB
Rust
//! # Utilities Module
|
||
//!
|
||
//! Common utilities for the data module including message parsing, validation,
|
||
//! performance monitoring, and helper functions for high-frequency trading.
|
||
//!
|
||
//! ## Features
|
||
//!
|
||
//! - Zero-copy message parsing for FIX and binary protocols
|
||
//! - High-precision timestamp utilities with RDTSC support
|
||
//! - Data validation and sanitization
|
||
//! - Performance monitoring and metrics collection
|
||
//! - Lock-free data structures for concurrent access
|
||
|
||
use crate::error::{DataError, Result};
|
||
use chrono::{DateTime, Utc};
|
||
use serde::{Deserialize, Serialize};
|
||
use std::collections::HashMap;
|
||
use std::sync::atomic::{AtomicU64, Ordering};
|
||
use std::sync::Arc;
|
||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||
use tracing::{error, warn};
|
||
|
||
/// Format timestamp as ISO 8601 string
|
||
pub fn format_timestamp(timestamp: DateTime<Utc>) -> String {
|
||
timestamp.to_rfc3339()
|
||
}
|
||
|
||
/// Calculate percentage change between two values
|
||
pub fn calculate_percentage_change(old_value: f64, new_value: f64) -> f64 {
|
||
if old_value == 0.0 {
|
||
return 0.0;
|
||
}
|
||
((new_value - old_value) / old_value) * 100.0
|
||
}
|
||
|
||
/// Normalize symbol to uppercase and remove invalid characters
|
||
pub fn normalize_symbol(symbol: &str) -> String {
|
||
symbol
|
||
.to_uppercase()
|
||
.chars()
|
||
.filter(|c| c.is_ascii_alphanumeric() || *c == '.' || *c == '-')
|
||
.collect()
|
||
}
|
||
|
||
/// High-precision timestamp utilities
|
||
pub mod timestamp {
|
||
use super::*;
|
||
use std::arch::x86_64::_rdtsc;
|
||
|
||
/// High-precision timestamp with nanosecond resolution
|
||
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
|
||
pub struct Timestamp {
|
||
pub nanos: u64,
|
||
}
|
||
|
||
impl Timestamp {
|
||
/// Create timestamp from current time
|
||
pub fn now() -> Self {
|
||
let now = SystemTime::now()
|
||
.duration_since(UNIX_EPOCH)
|
||
.unwrap_or_default();
|
||
Self {
|
||
nanos: now.as_nanos() as u64,
|
||
}
|
||
}
|
||
|
||
/// Create timestamp from RDTSC (CPU time stamp counter)
|
||
/// WARNING: Only use on systems with invariant TSC
|
||
pub fn from_rdtsc() -> Self {
|
||
// SAFETY: RDTSC instruction only reads CPU timestamp counter with no side effects
|
||
unsafe {
|
||
let tsc = _rdtsc();
|
||
// Convert TSC to nanoseconds (assumes 2.4 GHz CPU)
|
||
// In production, calibrate this conversion factor
|
||
let nanos = (tsc as f64 * 0.416667) as u64; // 1/(2.4*10^9) * 10^9
|
||
Self { nanos }
|
||
}
|
||
}
|
||
|
||
/// Create timestamp from chrono DateTime
|
||
pub fn from_datetime(dt: DateTime<Utc>) -> Self {
|
||
Self {
|
||
nanos: dt.timestamp_nanos_opt().unwrap_or(0) as u64,
|
||
}
|
||
}
|
||
|
||
/// Convert to chrono DateTime
|
||
pub fn to_datetime(self) -> DateTime<Utc> {
|
||
DateTime::from_timestamp_nanos(self.nanos as i64)
|
||
}
|
||
|
||
/// Get microseconds since Unix epoch
|
||
pub fn as_micros(self) -> u64 {
|
||
self.nanos / 1000
|
||
}
|
||
|
||
/// Get milliseconds since Unix epoch
|
||
pub fn as_millis(self) -> u64 {
|
||
self.nanos / 1_000_000
|
||
}
|
||
|
||
/// Calculate duration since another timestamp
|
||
pub fn duration_since(self, other: Timestamp) -> Duration {
|
||
if self.nanos >= other.nanos {
|
||
Duration::from_nanos(self.nanos - other.nanos)
|
||
} else {
|
||
Duration::from_nanos(0)
|
||
}
|
||
}
|
||
}
|
||
|
||
impl From<DateTime<Utc>> for Timestamp {
|
||
fn from(dt: DateTime<Utc>) -> Self {
|
||
Self::from_datetime(dt)
|
||
}
|
||
}
|
||
|
||
impl From<Timestamp> for DateTime<Utc> {
|
||
fn from(ts: Timestamp) -> Self {
|
||
ts.to_datetime()
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Message parsing utilities for `FIX` and binary protocols
|
||
pub mod parsing {
|
||
use super::*;
|
||
|
||
/// Zero-copy `FIX` message parser
|
||
pub struct FixParser {
|
||
soh: u8, // Start of Header character (0x01)
|
||
}
|
||
|
||
impl FixParser {
|
||
pub fn new() -> Self {
|
||
Self { soh: 0x01 }
|
||
}
|
||
|
||
/// Parse `FIX` message into field map
|
||
pub fn parse(&self, message: &str) -> Result<HashMap<u32, String>> {
|
||
let mut fields = HashMap::new();
|
||
|
||
for field in message.split(char::from(self.soh)) {
|
||
if field.is_empty() {
|
||
continue;
|
||
}
|
||
|
||
if let Some(eq_pos) = field.find('=') {
|
||
let tag_str = &field[..eq_pos];
|
||
let value = &field[eq_pos + 1..];
|
||
|
||
if let Ok(tag) = tag_str.parse::<u32>() {
|
||
fields.insert(tag, value.to_string());
|
||
}
|
||
}
|
||
}
|
||
|
||
Ok(fields)
|
||
}
|
||
|
||
/// Get field value by tag
|
||
pub fn get_field<'a>(
|
||
&self,
|
||
fields: &'a HashMap<u32, String>,
|
||
tag: u32,
|
||
) -> Option<&'a String> {
|
||
fields.get(&tag)
|
||
}
|
||
|
||
/// Get required field value by tag
|
||
pub fn get_required_field<'a>(
|
||
&self,
|
||
fields: &'a HashMap<u32, String>,
|
||
tag: u32,
|
||
) -> Result<&'a String> {
|
||
fields.get(&tag).ok_or_else(|| DataError::Parse {
|
||
message: format!("Required FIX field {} not found", tag),
|
||
})
|
||
}
|
||
|
||
/// Calculate `FIX` checksum
|
||
pub fn calculate_checksum(&self, message: &str) -> u8 {
|
||
message.bytes().fold(0_u8, |acc, b| acc.wrapping_add(b))
|
||
}
|
||
|
||
/// Validate `FIX` message checksum
|
||
pub fn validate_checksum(&self, message: &str) -> Result<bool> {
|
||
let fields = self.parse(message)?;
|
||
|
||
if let Some(checksum_str) = fields.get(&10) {
|
||
// Tag 10 = CheckSum
|
||
if let Ok(expected_checksum) = checksum_str.parse::<u8>() {
|
||
// Find the position of the checksum field
|
||
if let Some(checksum_pos) = message.rfind("10=") {
|
||
let message_without_checksum = &message[..checksum_pos];
|
||
let calculated = self.calculate_checksum(message_without_checksum);
|
||
return Ok(calculated == expected_checksum);
|
||
}
|
||
}
|
||
}
|
||
|
||
Err(DataError::Parse {
|
||
message: "Invalid or missing checksum field".to_string(),
|
||
})
|
||
}
|
||
}
|
||
|
||
impl Default for FixParser {
|
||
fn default() -> Self {
|
||
Self::new()
|
||
}
|
||
}
|
||
|
||
/// Binary message parser for efficient protocol handling
|
||
pub struct BinaryParser {
|
||
endianness: Endianness,
|
||
}
|
||
|
||
#[derive(Debug, Clone, Copy)]
|
||
pub enum Endianness {
|
||
BigEndian,
|
||
LittleEndian,
|
||
}
|
||
|
||
impl BinaryParser {
|
||
pub fn new(endianness: Endianness) -> Self {
|
||
Self { endianness }
|
||
}
|
||
|
||
/// Read u32 from bytes
|
||
pub fn read_u32(&self, bytes: &[u8], offset: usize) -> Result<u32> {
|
||
if bytes.len() < offset + 4 {
|
||
return Err(DataError::Parse {
|
||
message: "Insufficient bytes for u32".to_string(),
|
||
});
|
||
}
|
||
|
||
let value = match self.endianness {
|
||
Endianness::BigEndian => u32::from_be_bytes([
|
||
bytes[offset],
|
||
bytes[offset + 1],
|
||
bytes[offset + 2],
|
||
bytes[offset + 3],
|
||
]),
|
||
Endianness::LittleEndian => u32::from_le_bytes([
|
||
bytes[offset],
|
||
bytes[offset + 1],
|
||
bytes[offset + 2],
|
||
bytes[offset + 3],
|
||
]),
|
||
};
|
||
|
||
Ok(value)
|
||
}
|
||
|
||
/// Read u64 from bytes
|
||
pub fn read_u64(&self, bytes: &[u8], offset: usize) -> Result<u64> {
|
||
if bytes.len() < offset + 8 {
|
||
return Err(DataError::Parse {
|
||
message: "Insufficient bytes for u64".to_string(),
|
||
});
|
||
}
|
||
|
||
let mut array = [0_u8; 8];
|
||
array.copy_from_slice(&bytes[offset..offset + 8]);
|
||
|
||
let value = match self.endianness {
|
||
Endianness::BigEndian => u64::from_be_bytes(array),
|
||
Endianness::LittleEndian => u64::from_le_bytes(array),
|
||
};
|
||
|
||
Ok(value)
|
||
}
|
||
|
||
/// Read f64 from bytes
|
||
pub fn read_f64(&self, bytes: &[u8], offset: usize) -> Result<f64> {
|
||
let bits = self.read_u64(bytes, offset)?;
|
||
Ok(f64::from_bits(bits))
|
||
}
|
||
|
||
/// Read string with length prefix
|
||
pub fn read_string(&self, bytes: &[u8], offset: usize) -> Result<(String, usize)> {
|
||
let length = self.read_u32(bytes, offset)? as usize;
|
||
let start = offset + 4;
|
||
|
||
if bytes.len() < start + length {
|
||
return Err(DataError::Parse {
|
||
message: "Insufficient bytes for string".to_string(),
|
||
});
|
||
}
|
||
|
||
let string = String::from_utf8_lossy(&bytes[start..start + length]).to_string();
|
||
Ok((string, start + length))
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Data validation utilities
|
||
pub mod validation {
|
||
use super::*;
|
||
|
||
/// Data validator for market data events
|
||
pub struct DataValidator {
|
||
max_price_change: f64,
|
||
max_timestamp_skew: Duration,
|
||
enable_duplicate_detection: bool,
|
||
recent_events: HashMap<String, timestamp::Timestamp>,
|
||
}
|
||
|
||
impl DataValidator {
|
||
pub fn new(
|
||
max_price_change: f64,
|
||
max_timestamp_skew: Duration,
|
||
enable_duplicate_detection: bool,
|
||
) -> Self {
|
||
Self {
|
||
max_price_change,
|
||
max_timestamp_skew,
|
||
enable_duplicate_detection,
|
||
recent_events: HashMap::new(),
|
||
}
|
||
}
|
||
|
||
/// Validate price change percentage
|
||
pub fn validate_price_change(&self, old_price: f64, new_price: f64) -> Result<()> {
|
||
if old_price <= 0.0 || new_price <= 0.0 {
|
||
return Err(DataError::Validation {
|
||
field: "price".to_string(),
|
||
message: "Price must be positive".to_string(),
|
||
});
|
||
}
|
||
|
||
let change_percent = ((new_price - old_price) / old_price).abs() * 100.0;
|
||
if change_percent > self.max_price_change {
|
||
return Err(DataError::Validation {
|
||
field: "price_change".to_string(),
|
||
message: format!(
|
||
"Price change {:.2}% exceeds maximum {:.2}%",
|
||
change_percent, self.max_price_change
|
||
),
|
||
});
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Validate timestamp is within acceptable range
|
||
pub fn validate_timestamp(&self, timestamp: timestamp::Timestamp) -> Result<()> {
|
||
let now = timestamp::Timestamp::now();
|
||
let age = now.duration_since(timestamp);
|
||
|
||
if age > self.max_timestamp_skew {
|
||
return Err(DataError::Validation {
|
||
field: "timestamp".to_string(),
|
||
message: format!(
|
||
"Timestamp is {:.2}ms old, exceeds maximum {:.2}ms",
|
||
age.as_millis(),
|
||
self.max_timestamp_skew.as_millis()
|
||
),
|
||
});
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
|
||
/// Check for duplicate events
|
||
pub fn check_duplicate(
|
||
&mut self,
|
||
event_id: &str,
|
||
timestamp: timestamp::Timestamp,
|
||
) -> Result<()> {
|
||
if !self.enable_duplicate_detection {
|
||
return Ok(());
|
||
}
|
||
|
||
if let Some(&last_timestamp) = self.recent_events.get(event_id) {
|
||
if timestamp.nanos <= last_timestamp.nanos {
|
||
return Err(DataError::Validation {
|
||
field: "duplicate".to_string(),
|
||
message: format!("Duplicate or out-of-order event: {}", event_id),
|
||
});
|
||
}
|
||
}
|
||
|
||
self.recent_events.insert(event_id.to_string(), timestamp);
|
||
Ok(())
|
||
}
|
||
|
||
/// Validate symbol format
|
||
pub fn validate_symbol(&self, symbol: &str) -> Result<()> {
|
||
if symbol.is_empty() {
|
||
return Err(DataError::Validation {
|
||
field: "symbol".to_string(),
|
||
message: "Symbol cannot be empty".to_string(),
|
||
});
|
||
}
|
||
|
||
if symbol.len() > 12 {
|
||
return Err(DataError::Validation {
|
||
field: "symbol".to_string(),
|
||
message: "Symbol too long (max 12 characters)".to_string(),
|
||
});
|
||
}
|
||
|
||
if !symbol
|
||
.chars()
|
||
.all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
|
||
{
|
||
return Err(DataError::Validation {
|
||
field: "symbol".to_string(),
|
||
message: "Symbol contains invalid characters".to_string(),
|
||
});
|
||
}
|
||
|
||
Ok(())
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Performance monitoring utilities
|
||
pub mod monitoring {
|
||
use super::*;
|
||
|
||
/// Performance metrics collector
|
||
#[derive(Debug, Clone)]
|
||
pub struct MetricsCollector {
|
||
counters: Arc<parking_lot::RwLock<HashMap<String, AtomicU64>>>,
|
||
gauges: Arc<parking_lot::RwLock<HashMap<String, AtomicU64>>>,
|
||
histograms: Arc<parking_lot::RwLock<HashMap<String, Histogram>>>,
|
||
}
|
||
|
||
impl MetricsCollector {
|
||
pub fn new() -> Self {
|
||
Self {
|
||
counters: Arc::new(parking_lot::RwLock::new(HashMap::new())),
|
||
gauges: Arc::new(parking_lot::RwLock::new(HashMap::new())),
|
||
histograms: Arc::new(parking_lot::RwLock::new(HashMap::new())),
|
||
}
|
||
}
|
||
|
||
/// Increment counter
|
||
pub fn increment_counter(&self, name: &str, value: u64) {
|
||
let counters = self.counters.read();
|
||
if let Some(counter) = counters.get(name) {
|
||
counter.fetch_add(value, Ordering::Relaxed);
|
||
} else {
|
||
drop(counters);
|
||
let mut counters = self.counters.write();
|
||
counters
|
||
.entry(name.to_string())
|
||
.or_insert_with(|| AtomicU64::new(0))
|
||
.fetch_add(value, Ordering::Relaxed);
|
||
}
|
||
}
|
||
|
||
/// Set gauge value
|
||
pub fn set_gauge(&self, name: &str, value: u64) {
|
||
let gauges = self.gauges.read();
|
||
if let Some(gauge) = gauges.get(name) {
|
||
gauge.store(value, Ordering::Relaxed);
|
||
} else {
|
||
drop(gauges);
|
||
let mut gauges = self.gauges.write();
|
||
gauges
|
||
.entry(name.to_string())
|
||
.or_insert_with(|| AtomicU64::new(0))
|
||
.store(value, Ordering::Relaxed);
|
||
}
|
||
}
|
||
|
||
/// Record histogram value
|
||
pub fn record_histogram(&self, name: &str, value: f64) {
|
||
let mut histograms = self.histograms.write();
|
||
histograms
|
||
.entry(name.to_string())
|
||
.or_insert_with(Histogram::new)
|
||
.record(value);
|
||
}
|
||
|
||
/// Get counter value
|
||
pub fn get_counter(&self, name: &str) -> u64 {
|
||
self.counters
|
||
.read()
|
||
.get(name)
|
||
.map(|counter| counter.load(Ordering::Relaxed))
|
||
.unwrap_or(0)
|
||
}
|
||
|
||
/// Get gauge value
|
||
pub fn get_gauge(&self, name: &str) -> u64 {
|
||
self.gauges
|
||
.read()
|
||
.get(name)
|
||
.map(|gauge| gauge.load(Ordering::Relaxed))
|
||
.unwrap_or(0)
|
||
}
|
||
|
||
/// Get histogram statistics
|
||
pub fn get_histogram_stats(&self, name: &str) -> Option<HistogramStats> {
|
||
self.histograms.read().get(name).map(|h| h.stats())
|
||
}
|
||
|
||
/// Export all metrics
|
||
pub fn export_metrics(&self) -> MetricsSnapshot {
|
||
MetricsSnapshot {
|
||
counters: self
|
||
.counters
|
||
.read()
|
||
.iter()
|
||
.map(|(k, v)| (k.clone(), v.load(Ordering::Relaxed)))
|
||
.collect(),
|
||
gauges: self
|
||
.gauges
|
||
.read()
|
||
.iter()
|
||
.map(|(k, v)| (k.clone(), v.load(Ordering::Relaxed)))
|
||
.collect(),
|
||
histograms: self
|
||
.histograms
|
||
.read()
|
||
.iter()
|
||
.map(|(k, v)| (k.clone(), v.stats()))
|
||
.collect(),
|
||
timestamp: timestamp::Timestamp::now(),
|
||
}
|
||
}
|
||
}
|
||
|
||
impl Default for MetricsCollector {
|
||
fn default() -> Self {
|
||
Self::new()
|
||
}
|
||
}
|
||
|
||
/// Histogram for recording value distributions
|
||
#[derive(Debug, Clone)]
|
||
pub struct Histogram {
|
||
values: Vec<f64>,
|
||
min: f64,
|
||
max: f64,
|
||
sum: f64,
|
||
count: u64,
|
||
}
|
||
|
||
impl Histogram {
|
||
pub fn new() -> Self {
|
||
Self {
|
||
values: Vec::new(),
|
||
min: f64::INFINITY,
|
||
max: f64::NEG_INFINITY,
|
||
sum: 0.0,
|
||
count: 0,
|
||
}
|
||
}
|
||
|
||
pub fn record(&mut self, value: f64) {
|
||
self.values.push(value);
|
||
self.min = self.min.min(value);
|
||
self.max = self.max.max(value);
|
||
self.sum += value;
|
||
self.count += 1;
|
||
}
|
||
|
||
pub fn stats(&self) -> HistogramStats {
|
||
if self.count == 0 {
|
||
return HistogramStats::default();
|
||
}
|
||
|
||
let mean = self.sum / self.count as f64;
|
||
|
||
// Calculate percentiles
|
||
let mut sorted = self.values.clone();
|
||
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
|
||
|
||
let p50 = percentile(&sorted, 0.5);
|
||
let p95 = percentile(&sorted, 0.95);
|
||
let p99 = percentile(&sorted, 0.99);
|
||
|
||
HistogramStats {
|
||
count: self.count,
|
||
min: self.min,
|
||
max: self.max,
|
||
mean,
|
||
p50,
|
||
p95,
|
||
p99,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Calculate percentile from sorted values using linear interpolation
|
||
fn percentile(sorted_values: &[f64], p: f64) -> f64 {
|
||
if sorted_values.is_empty() {
|
||
return 0.0;
|
||
}
|
||
|
||
if sorted_values.len() == 1 {
|
||
return sorted_values[0];
|
||
}
|
||
|
||
// Use linear interpolation for more accurate percentile calculation
|
||
let index = p * (sorted_values.len() - 1) as f64;
|
||
let lower_index = index.floor() as usize;
|
||
let upper_index = index.ceil() as usize;
|
||
|
||
// SAFETY: lower_index and upper_index are guaranteed to be valid:
|
||
// - index = p * (len - 1) where 0 <= p <= 1
|
||
// - lower_index = floor(index) <= index <= len - 1
|
||
// - upper_index = ceil(index) <= index <= len - 1
|
||
#[allow(clippy::indexing_slicing)]
|
||
if lower_index == upper_index {
|
||
sorted_values[lower_index]
|
||
} else {
|
||
let lower_value = sorted_values[lower_index];
|
||
let upper_value = sorted_values[upper_index];
|
||
let fraction = index - lower_index as f64;
|
||
lower_value + (upper_value - lower_value) * fraction
|
||
}
|
||
}
|
||
|
||
/// Histogram statistics
|
||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||
pub struct HistogramStats {
|
||
pub count: u64,
|
||
pub min: f64,
|
||
pub max: f64,
|
||
pub mean: f64,
|
||
pub p50: f64,
|
||
pub p95: f64,
|
||
pub p99: f64,
|
||
}
|
||
|
||
impl Default for HistogramStats {
|
||
fn default() -> Self {
|
||
Self {
|
||
count: 0,
|
||
min: 0.0,
|
||
max: 0.0,
|
||
mean: 0.0,
|
||
p50: 0.0,
|
||
p95: 0.0,
|
||
p99: 0.0,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Snapshot of all metrics at a point in time
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct MetricsSnapshot {
|
||
pub counters: HashMap<String, u64>,
|
||
pub gauges: HashMap<String, u64>,
|
||
pub histograms: HashMap<String, HistogramStats>,
|
||
pub timestamp: timestamp::Timestamp,
|
||
}
|
||
|
||
/// Latency measurement utility
|
||
pub struct LatencyMeasurer {
|
||
start_time: timestamp::Timestamp,
|
||
name: String,
|
||
metrics: MetricsCollector,
|
||
}
|
||
|
||
impl LatencyMeasurer {
|
||
pub fn start(name: String, metrics: MetricsCollector) -> Self {
|
||
Self {
|
||
start_time: timestamp::Timestamp::now(),
|
||
name,
|
||
metrics,
|
||
}
|
||
}
|
||
|
||
pub fn finish(self) -> Duration {
|
||
let end_time = timestamp::Timestamp::now();
|
||
let duration = end_time.duration_since(self.start_time);
|
||
|
||
// Record latency in histogram
|
||
self.metrics.record_histogram(
|
||
&format!("{}_latency_us", self.name),
|
||
duration.as_micros() as f64,
|
||
);
|
||
|
||
duration
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Lock-free data structures for concurrent access
|
||
pub mod lockfree {
|
||
use super::*;
|
||
use crossbeam::queue::SegQueue;
|
||
use std::sync::atomic::{AtomicBool, AtomicUsize};
|
||
|
||
/// Lock-free message queue for high-frequency trading
|
||
pub struct LockFreeQueue<T> {
|
||
queue: SegQueue<T>,
|
||
size: AtomicUsize,
|
||
max_size: usize,
|
||
overflow: AtomicBool,
|
||
}
|
||
|
||
impl<T> LockFreeQueue<T> {
|
||
pub fn new(max_size: usize) -> Self {
|
||
Self {
|
||
queue: SegQueue::new(),
|
||
size: AtomicUsize::new(0),
|
||
max_size,
|
||
overflow: AtomicBool::new(false),
|
||
}
|
||
}
|
||
|
||
/// Push item to queue (non-blocking)
|
||
pub fn push(&self, item: T) -> bool {
|
||
let current_size = self.size.load(Ordering::Relaxed);
|
||
|
||
if current_size >= self.max_size {
|
||
self.overflow.store(true, Ordering::Relaxed);
|
||
return false;
|
||
}
|
||
|
||
self.queue.push(item);
|
||
self.size.fetch_add(1, Ordering::Relaxed);
|
||
true
|
||
}
|
||
|
||
/// Pop item from queue (non-blocking)
|
||
pub fn pop(&self) -> Option<T> {
|
||
match self.queue.pop() {
|
||
Some(item) => {
|
||
self.size.fetch_sub(1, Ordering::Relaxed);
|
||
Some(item)
|
||
},
|
||
None => None,
|
||
}
|
||
}
|
||
|
||
/// Get current queue size
|
||
pub fn len(&self) -> usize {
|
||
self.size.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// Check if queue is empty
|
||
pub fn is_empty(&self) -> bool {
|
||
self.len() == 0
|
||
}
|
||
|
||
/// Check if queue has overflowed
|
||
pub fn has_overflowed(&self) -> bool {
|
||
self.overflow.load(Ordering::Relaxed)
|
||
}
|
||
|
||
/// Reset overflow flag
|
||
pub fn reset_overflow(&self) {
|
||
self.overflow.store(false, Ordering::Relaxed);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Network utilities for broker connections
|
||
pub mod network {
|
||
use super::*;
|
||
use tokio::time::{sleep, timeout};
|
||
|
||
/// Connection helper with automatic retry
|
||
pub struct ConnectionHelper {
|
||
max_attempts: u32,
|
||
initial_delay: Duration,
|
||
max_delay: Duration,
|
||
backoff_multiplier: f64,
|
||
jitter_factor: f64,
|
||
}
|
||
|
||
impl ConnectionHelper {
|
||
pub fn new(
|
||
max_attempts: u32,
|
||
initial_delay: Duration,
|
||
max_delay: Duration,
|
||
backoff_multiplier: f64,
|
||
jitter_factor: f64,
|
||
) -> Self {
|
||
Self {
|
||
max_attempts,
|
||
initial_delay,
|
||
max_delay,
|
||
backoff_multiplier,
|
||
jitter_factor,
|
||
}
|
||
}
|
||
|
||
/// Retry connection with exponential backoff
|
||
pub async fn retry_connect<F, Fut, T, E>(
|
||
&self,
|
||
mut connect_fn: F,
|
||
) -> std::result::Result<T, E>
|
||
where
|
||
F: FnMut() -> Fut,
|
||
Fut: std::future::Future<Output = std::result::Result<T, E>>,
|
||
E: std::fmt::Display,
|
||
{
|
||
let mut attempt = 0;
|
||
let mut delay = self.initial_delay;
|
||
|
||
loop {
|
||
attempt += 1;
|
||
|
||
match connect_fn().await {
|
||
Ok(result) => return Ok(result),
|
||
Err(e) => {
|
||
if attempt >= self.max_attempts {
|
||
error!("Connection failed after {} attempts: {}", attempt, e);
|
||
return Err(e);
|
||
}
|
||
|
||
warn!(
|
||
"Connection attempt {} failed: {}, retrying in {:?}",
|
||
attempt, e, delay
|
||
);
|
||
|
||
// Add jitter to prevent thundering herd
|
||
let jitter =
|
||
delay.as_millis() as f64 * self.jitter_factor * fastrand::f64();
|
||
let jittered_delay = delay + Duration::from_millis(jitter as u64);
|
||
|
||
sleep(jittered_delay).await;
|
||
|
||
// Exponential backoff
|
||
delay = Duration::from_millis(
|
||
((delay.as_millis() as f64 * self.backoff_multiplier) as u64)
|
||
.min(self.max_delay.as_millis() as u64),
|
||
);
|
||
},
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Connect with timeout
|
||
pub async fn connect_with_timeout<F, Fut, T, E>(
|
||
&self,
|
||
connect_fn: F,
|
||
timeout_duration: Duration,
|
||
) -> std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>
|
||
where
|
||
F: FnOnce() -> Fut,
|
||
Fut: std::future::Future<Output = std::result::Result<T, E>>,
|
||
E: std::error::Error + Send + Sync + 'static,
|
||
{
|
||
timeout(timeout_duration, connect_fn())
|
||
.await
|
||
.map_err(|_| {
|
||
Box::<dyn std::error::Error + Send + Sync>::from("Connection timeout")
|
||
})?
|
||
.map_err(|e| e.into())
|
||
}
|
||
}
|
||
|
||
impl Default for ConnectionHelper {
|
||
fn default() -> Self {
|
||
Self::new(
|
||
10, // max_attempts
|
||
Duration::from_millis(1000), // initial_delay
|
||
Duration::from_millis(60000), // max_delay
|
||
2.0, // backoff_multiplier
|
||
0.1, // jitter_factor
|
||
)
|
||
}
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn test_timestamp_creation() {
|
||
let ts1 = timestamp::Timestamp::now();
|
||
let ts2 = timestamp::Timestamp::now();
|
||
assert!(ts2.nanos >= ts1.nanos);
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser() {
|
||
let parser = parsing::FixParser::new();
|
||
let message = "8=FIX.4.4\x0135=D\x0149=SENDER\x0156=TARGET\x0110=123\x01";
|
||
let fields = parser.parse(message).unwrap();
|
||
|
||
assert_eq!(fields.get(&8), Some(&"FIX.4.4".to_string()));
|
||
assert_eq!(fields.get(&35), Some(&"D".to_string()));
|
||
}
|
||
|
||
#[test]
|
||
fn test_data_validator() {
|
||
let validator = validation::DataValidator::new(10.0, Duration::from_secs(60), true);
|
||
|
||
// Test price validation
|
||
assert!(validator.validate_price_change(100.0, 105.0).is_ok());
|
||
assert!(validator.validate_price_change(100.0, 120.0).is_err());
|
||
|
||
// Test symbol validation
|
||
assert!(validator.validate_symbol("AAPL").is_ok());
|
||
assert!(validator.validate_symbol("").is_err());
|
||
assert!(validator.validate_symbol("VERY_LONG_SYMBOL_NAME").is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_metrics_collector() {
|
||
let metrics = monitoring::MetricsCollector::new();
|
||
|
||
metrics.increment_counter("test_counter", 1);
|
||
metrics.set_gauge("test_gauge", 42);
|
||
metrics.record_histogram("test_histogram", 1.5);
|
||
|
||
assert_eq!(metrics.get_counter("test_counter"), 1);
|
||
assert_eq!(metrics.get_gauge("test_gauge"), 42);
|
||
|
||
let stats = metrics.get_histogram_stats("test_histogram").unwrap();
|
||
assert_eq!(stats.count, 1);
|
||
assert_eq!(stats.mean, 1.5);
|
||
}
|
||
|
||
#[test]
|
||
fn test_lockfree_queue() {
|
||
let queue = lockfree::LockFreeQueue::new(10);
|
||
|
||
assert!(queue.push("item1"));
|
||
assert!(queue.push("item2"));
|
||
assert_eq!(queue.len(), 2);
|
||
|
||
assert_eq!(queue.pop(), Some("item1"));
|
||
assert_eq!(queue.pop(), Some("item2"));
|
||
assert!(queue.is_empty());
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_connection_helper() {
|
||
tokio::time::pause();
|
||
|
||
let helper = network::ConnectionHelper::default();
|
||
let mut attempts = 0;
|
||
|
||
let result = helper
|
||
.retry_connect(|| {
|
||
attempts += 1;
|
||
async move {
|
||
if attempts < 3 {
|
||
Err("Connection failed")
|
||
} else {
|
||
Ok("Connected")
|
||
}
|
||
}
|
||
})
|
||
.await;
|
||
|
||
assert_eq!(result, Ok("Connected"));
|
||
assert_eq!(attempts, 3);
|
||
}
|
||
|
||
// === COMPREHENSIVE TEST EXPANSION (50+ NEW TESTS) ===
|
||
use chrono::{DateTime, Utc};
|
||
|
||
// TIMESTAMP MODULE TESTS (10 new tests)
|
||
#[test]
|
||
fn test_timestamp_duration_edges() {
|
||
let earlier = timestamp::Timestamp::now();
|
||
std::thread::sleep(Duration::from_millis(1));
|
||
let later = timestamp::Timestamp::now();
|
||
|
||
// Happy-path: later – earlier > 0
|
||
let dur = later.duration_since(earlier);
|
||
assert!(dur.as_nanos() > 0, "expected positive duration");
|
||
|
||
// Underflow clamped to zero
|
||
let dur_under = earlier.duration_since(later);
|
||
assert_eq!(dur_under, Duration::from_nanos(0));
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_roundtrip_datetime() {
|
||
let ts = timestamp::Timestamp::now();
|
||
let dt = ts.to_datetime();
|
||
let ts2 = timestamp::Timestamp::from_datetime(dt);
|
||
assert_eq!(ts, ts2);
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_from_rdtsc() {
|
||
let ts1 = timestamp::Timestamp::from_rdtsc();
|
||
let ts2 = timestamp::Timestamp::from_rdtsc();
|
||
// RDTSC should be monotonic
|
||
assert!(ts2.nanos >= ts1.nanos);
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_conversions() {
|
||
let ts = timestamp::Timestamp {
|
||
nanos: 1_234_567_890_123,
|
||
};
|
||
assert_eq!(ts.as_micros(), 1_234_567_890);
|
||
assert_eq!(ts.as_millis(), 1_234_567);
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_overflow_protection() {
|
||
let max_ts = timestamp::Timestamp { nanos: u64::MAX };
|
||
let min_ts = timestamp::Timestamp { nanos: 0 };
|
||
|
||
// Should not panic on max values
|
||
let dur = max_ts.duration_since(min_ts);
|
||
assert_eq!(dur.as_nanos() as u64, u64::MAX);
|
||
|
||
// Reverse should clamp to zero
|
||
let dur_rev = min_ts.duration_since(max_ts);
|
||
assert_eq!(dur_rev, Duration::from_nanos(0));
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_from_traits() {
|
||
let dt = DateTime::from_timestamp_nanos(1_234_567_890_123_456_789);
|
||
let ts: timestamp::Timestamp = dt.into();
|
||
let dt_back: DateTime<Utc> = ts.into();
|
||
assert_eq!(dt, dt_back);
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_zero_edge_case() {
|
||
let zero_ts = timestamp::Timestamp { nanos: 0 };
|
||
assert_eq!(zero_ts.as_micros(), 0);
|
||
assert_eq!(zero_ts.as_millis(), 0);
|
||
|
||
// to_datetime with zero should not panic
|
||
let dt = zero_ts.to_datetime();
|
||
assert_eq!(dt.timestamp(), 0);
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_ordering() {
|
||
let ts1 = timestamp::Timestamp { nanos: 1000 };
|
||
let ts2 = timestamp::Timestamp { nanos: 2000 };
|
||
let ts3 = timestamp::Timestamp { nanos: 1000 };
|
||
|
||
assert!(ts2 > ts1);
|
||
assert!(ts1 < ts2);
|
||
assert_eq!(ts1, ts3);
|
||
assert!(ts1 <= ts3);
|
||
assert!(ts2 >= ts1);
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_serialization() {
|
||
let ts = timestamp::Timestamp {
|
||
nanos: 1_234_567_890,
|
||
};
|
||
let json = serde_json::to_string(&ts).unwrap();
|
||
let deserialized: timestamp::Timestamp = serde_json::from_str(&json).unwrap();
|
||
assert_eq!(ts, deserialized);
|
||
}
|
||
|
||
#[test]
|
||
fn test_timestamp_large_duration() {
|
||
let ts1 = timestamp::Timestamp {
|
||
nanos: 1_000_000_000,
|
||
}; // 1 second
|
||
let ts2 = timestamp::Timestamp {
|
||
nanos: 3_600_000_000_000,
|
||
}; // 1 hour
|
||
let dur = ts2.duration_since(ts1);
|
||
assert_eq!(dur.as_secs(), 3599); // ~1 hour
|
||
}
|
||
|
||
// FIX PARSER TESTS (12 new tests)
|
||
#[test]
|
||
fn test_fix_parser_checksum_paths() {
|
||
let parser = parsing::FixParser::new();
|
||
// Build simple FIX string: "8=FIX.4.4<SOH>10=CS<SOH>"
|
||
let body = "8=FIX.4.4\u{1}";
|
||
let checksum = parser.calculate_checksum(body);
|
||
let msg_ok = format!("{body}10={checksum}\u{1}");
|
||
assert!(parser.validate_checksum(&msg_ok).unwrap());
|
||
|
||
// Use a definitely wrong checksum (different from calculated)
|
||
let wrong_checksum = checksum.wrapping_add(1);
|
||
let msg_bad = format!("{body}10={wrong_checksum}\u{1}");
|
||
assert!(!parser.validate_checksum(&msg_bad).unwrap_or(true));
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_required_field_err() {
|
||
let parser = parsing::FixParser::new();
|
||
let fields = parser.parse("8=FIX.4.4\u{1}").unwrap();
|
||
let err = parser.get_required_field(&fields, 35); // tag 35 missing
|
||
assert!(err.is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_empty_message() {
|
||
let parser = parsing::FixParser::new();
|
||
let fields = parser.parse("").unwrap();
|
||
assert!(fields.is_empty());
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_malformed_fields() {
|
||
let parser = parsing::FixParser::new();
|
||
|
||
// No equals sign
|
||
let fields = parser.parse("8FIX.4.4\u{1}").unwrap();
|
||
assert!(fields.is_empty());
|
||
|
||
// Invalid tag (non-numeric)
|
||
let fields = parser.parse("abc=FIX.4.4\u{1}").unwrap();
|
||
assert!(fields.is_empty());
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_checksum_edge_cases() {
|
||
let parser = parsing::FixParser::new();
|
||
|
||
// Message without checksum
|
||
let result = parser.validate_checksum("8=FIX.4.4\u{1}");
|
||
assert!(result.is_err());
|
||
|
||
// Checksum with invalid format
|
||
let result = parser.validate_checksum("8=FIX.4.4\u{1}10=abc\u{1}");
|
||
assert!(result.is_err());
|
||
|
||
// Multiple checksums (should use last one) - this is malformed and should fail
|
||
let body = "8=FIX.4.4\u{1}10=999\u{1}";
|
||
let checksum = parser.calculate_checksum("8=FIX.4.4\u{1}10=999\u{1}");
|
||
let msg = format!("{body}10={checksum}\u{1}");
|
||
// This message has duplicate checksum fields which is malformed
|
||
// The parser will find the FIRST occurrence of "10=" when parsing
|
||
assert!(parser.validate_checksum(&msg).is_ok() || parser.validate_checksum(&msg).is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_wrapped_checksum() {
|
||
let parser = parsing::FixParser::new();
|
||
|
||
// Test checksum wrapping (sum > 255)
|
||
let long_body = "8=FIX.4.4\u{1}35=D\u{1}49=SENDER_WITH_VERY_LONG_NAME\u{1}56=TARGET_WITH_VERY_LONG_NAME\u{1}";
|
||
let checksum = parser.calculate_checksum(long_body);
|
||
let msg = format!("{long_body}10={:03}\u{1}", checksum);
|
||
assert!(parser.validate_checksum(&msg).unwrap());
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_default() {
|
||
let parser1 = parsing::FixParser::new();
|
||
let parser2 = parsing::FixParser::default();
|
||
|
||
let message = "8=FIX.4.4\u{1}35=D\u{1}";
|
||
let fields1 = parser1.parse(message).unwrap();
|
||
let fields2 = parser2.parse(message).unwrap();
|
||
assert_eq!(fields1, fields2);
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_special_characters() {
|
||
let parser = parsing::FixParser::new();
|
||
|
||
// Field with special characters
|
||
let message = "8=FIX.4.4\u{1}58=Special: @#$%^&*()\u{1}";
|
||
let fields = parser.parse(message).unwrap();
|
||
assert_eq!(fields.get(&58), Some(&"Special: @#$%^&*()".to_string()));
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_zero_checksum() {
|
||
let parser = parsing::FixParser::new();
|
||
|
||
// Create a message that results in checksum 0
|
||
let test_bytes = vec![0u8; 256]; // Sum = 0 after wrapping
|
||
let test_str = String::from_utf8_lossy(&test_bytes);
|
||
let checksum = parser.calculate_checksum(&test_str);
|
||
assert_eq!(checksum, 0);
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_large_tag_numbers() {
|
||
let parser = parsing::FixParser::new();
|
||
|
||
let message = "9999999=TestValue\u{1}";
|
||
let fields = parser.parse(message).unwrap();
|
||
assert_eq!(fields.get(&9999999), Some(&"TestValue".to_string()));
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_consecutive_soh() {
|
||
let parser = parsing::FixParser::new();
|
||
|
||
// Multiple consecutive SOH characters
|
||
let message = "8=FIX.4.4\u{1}\u{1}\u{1}35=D\u{1}";
|
||
let fields = parser.parse(message).unwrap();
|
||
assert_eq!(fields.len(), 2);
|
||
assert_eq!(fields.get(&8), Some(&"FIX.4.4".to_string()));
|
||
assert_eq!(fields.get(&35), Some(&"D".to_string()));
|
||
}
|
||
|
||
#[test]
|
||
fn test_fix_parser_equals_in_value() {
|
||
let parser = parsing::FixParser::new();
|
||
|
||
// Value contains equals sign
|
||
let message = "58=Math: 2+2=4\u{1}";
|
||
let fields = parser.parse(message).unwrap();
|
||
assert_eq!(fields.get(&58), Some(&"Math: 2+2=4".to_string()));
|
||
}
|
||
|
||
// BINARY PARSER TESTS (8 new tests)
|
||
#[test]
|
||
fn test_binary_parser_u32_and_string() {
|
||
use parsing::{BinaryParser, Endianness};
|
||
|
||
// Big-endian u32 = 0x01020304
|
||
let bytes = [1u8, 2, 3, 4];
|
||
let parser_be = BinaryParser::new(Endianness::BigEndian);
|
||
assert_eq!(parser_be.read_u32(&bytes, 0).unwrap(), 0x01020304);
|
||
|
||
// Little-endian u32 = 0x04030201
|
||
let parser_le = BinaryParser::new(Endianness::LittleEndian);
|
||
assert_eq!(parser_le.read_u32(&bytes, 0).unwrap(), 0x04030201);
|
||
|
||
// Insufficient bytes
|
||
assert!(parser_be.read_u32(&bytes[..3], 0).is_err());
|
||
|
||
// Length-prefixed string "ABC"
|
||
let mut data = Vec::new();
|
||
data.extend_from_slice(&3u32.to_le_bytes()); // length prefix
|
||
data.extend_from_slice(b"ABC");
|
||
let (s, next) = parser_le.read_string(&data, 0).unwrap();
|
||
assert_eq!(s, "ABC");
|
||
assert_eq!(next, data.len());
|
||
}
|
||
|
||
#[test]
|
||
fn test_binary_parser_u64() {
|
||
use parsing::{BinaryParser, Endianness};
|
||
|
||
let bytes = [0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08];
|
||
let parser_be = BinaryParser::new(Endianness::BigEndian);
|
||
let parser_le = BinaryParser::new(Endianness::LittleEndian);
|
||
|
||
let value_be = parser_be.read_u64(&bytes, 0).unwrap();
|
||
let value_le = parser_le.read_u64(&bytes, 0).unwrap();
|
||
|
||
assert_eq!(value_be, 0x0102030405060708);
|
||
assert_eq!(value_le, 0x0807060504030201);
|
||
|
||
// Insufficient bytes
|
||
assert!(parser_be.read_u64(&bytes[..7], 0).is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_binary_parser_f64() {
|
||
use parsing::{BinaryParser, Endianness};
|
||
|
||
let value = 3.14159265359_f64;
|
||
let bits = value.to_bits();
|
||
let bytes = bits.to_le_bytes();
|
||
|
||
let parser_le = BinaryParser::new(Endianness::LittleEndian);
|
||
let parsed = parser_le.read_f64(&bytes, 0).unwrap();
|
||
|
||
assert!((parsed - value).abs() < f64::EPSILON);
|
||
}
|
||
|
||
#[test]
|
||
fn test_binary_parser_string_edge_cases() {
|
||
use parsing::{BinaryParser, Endianness};
|
||
|
||
let parser_le = BinaryParser::new(Endianness::LittleEndian);
|
||
|
||
// Empty string
|
||
let mut data = Vec::new();
|
||
data.extend_from_slice(&0u32.to_le_bytes());
|
||
let (s, next) = parser_le.read_string(&data, 0).unwrap();
|
||
assert_eq!(s, "");
|
||
assert_eq!(next, 4);
|
||
|
||
// String with Unicode
|
||
let unicode_str = "Hello 世界";
|
||
let utf8_bytes = unicode_str.as_bytes();
|
||
let mut data = Vec::new();
|
||
data.extend_from_slice(&(utf8_bytes.len() as u32).to_le_bytes());
|
||
data.extend_from_slice(utf8_bytes);
|
||
let (s, next) = parser_le.read_string(&data, 0).unwrap();
|
||
assert_eq!(s, unicode_str);
|
||
assert_eq!(next, 4 + utf8_bytes.len());
|
||
}
|
||
|
||
#[test]
|
||
fn test_binary_parser_offset_bounds() {
|
||
use parsing::{BinaryParser, Endianness};
|
||
|
||
let bytes = [1u8, 2, 3, 4, 5, 6, 7, 8];
|
||
let parser = BinaryParser::new(Endianness::BigEndian);
|
||
|
||
// Valid offset
|
||
assert_eq!(parser.read_u32(&bytes, 4).unwrap(), 0x05060708);
|
||
|
||
// Invalid offset (would read past end)
|
||
assert!(parser.read_u32(&bytes, 5).is_err());
|
||
assert!(parser.read_u32(&bytes, 8).is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_binary_parser_string_length_overflow() {
|
||
use parsing::{BinaryParser, Endianness};
|
||
|
||
let parser = BinaryParser::new(Endianness::LittleEndian);
|
||
|
||
// Length prefix larger than remaining data
|
||
let mut data = Vec::new();
|
||
data.extend_from_slice(&1000u32.to_le_bytes()); // claims 1000 bytes
|
||
data.extend_from_slice(b"short"); // but only 5 bytes available
|
||
|
||
let result = parser.read_string(&data, 0);
|
||
assert!(result.is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_binary_parser_invalid_utf8() {
|
||
use parsing::{BinaryParser, Endianness};
|
||
|
||
let parser = BinaryParser::new(Endianness::LittleEndian);
|
||
|
||
// Invalid UTF-8 sequence
|
||
let invalid_utf8 = [0xFF, 0xFE, 0xFD];
|
||
let mut data = Vec::new();
|
||
data.extend_from_slice(&(invalid_utf8.len() as u32).to_le_bytes());
|
||
data.extend_from_slice(&invalid_utf8);
|
||
|
||
// Should not panic, but use lossy conversion
|
||
let (s, _) = parser.read_string(&data, 0).unwrap();
|
||
assert!(!s.is_empty()); // Should contain replacement characters
|
||
}
|
||
|
||
#[test]
|
||
fn test_binary_parser_zero_offset() {
|
||
use parsing::{BinaryParser, Endianness};
|
||
|
||
let bytes = [0x12, 0x34, 0x56, 0x78];
|
||
let parser = BinaryParser::new(Endianness::BigEndian);
|
||
|
||
assert_eq!(parser.read_u32(&bytes, 0).unwrap(), 0x12345678);
|
||
}
|
||
|
||
// VALIDATION TESTS (10 new tests)
|
||
#[test]
|
||
fn test_data_validator_error_paths() {
|
||
let mut v = validation::DataValidator::new(5.0, Duration::from_secs(1), true);
|
||
|
||
// Too large price change
|
||
assert!(v.validate_price_change(100.0, 120.0).is_err());
|
||
|
||
// Just within limit should pass
|
||
assert!(v.validate_price_change(100.0, 105.0).is_ok());
|
||
|
||
// Negative price
|
||
assert!(v.validate_price_change(-1.0, 1.0).is_err());
|
||
assert!(v.validate_price_change(1.0, -1.0).is_err());
|
||
assert!(v.validate_price_change(0.0, 1.0).is_err());
|
||
|
||
// Timestamp skew
|
||
let old_ts = timestamp::Timestamp {
|
||
nanos: timestamp::Timestamp::now()
|
||
.nanos
|
||
.saturating_sub(2_000_000_000), // 2s ago
|
||
};
|
||
assert!(v.validate_timestamp(old_ts).is_err());
|
||
|
||
// Valid recent timestamp
|
||
let recent_ts = timestamp::Timestamp::now();
|
||
assert!(v.validate_timestamp(recent_ts).is_ok());
|
||
|
||
// Duplicate / out-of-order event
|
||
let id = "EVT1";
|
||
let ts1 = timestamp::Timestamp::now();
|
||
let ts2 = ts1; // equal timestamp considered duplicate/out-of-order
|
||
v.check_duplicate(id, ts1).unwrap();
|
||
assert!(v.check_duplicate(id, ts2).is_err());
|
||
|
||
// Invalid symbol characters
|
||
assert!(v.validate_symbol("BAD!SYM").is_err());
|
||
assert!(v.validate_symbol("").is_err());
|
||
assert!(v.validate_symbol("VERY_LONG_SYMBOL_NAME").is_err());
|
||
|
||
// Valid symbols
|
||
assert!(v.validate_symbol("AAPL").is_ok());
|
||
assert!(v.validate_symbol("BRK.A").is_ok());
|
||
assert!(v.validate_symbol("BTC-USD").is_ok());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_price_change_edge_cases() {
|
||
let validator = validation::DataValidator::new(10.0, Duration::from_secs(60), false);
|
||
|
||
// Exactly at limit
|
||
assert!(validator.validate_price_change(100.0, 110.0).is_ok());
|
||
assert!(validator.validate_price_change(100.0, 90.0).is_ok());
|
||
|
||
// Just over limit
|
||
assert!(validator.validate_price_change(100.0, 110.1).is_err());
|
||
assert!(validator.validate_price_change(100.0, 89.9).is_err());
|
||
|
||
// Very small prices
|
||
assert!(validator.validate_price_change(0.0001, 0.00011).is_ok());
|
||
|
||
// Large prices
|
||
assert!(validator.validate_price_change(10000.0, 11000.0).is_ok());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_duplicate_detection_disabled() {
|
||
let mut validator = validation::DataValidator::new(10.0, Duration::from_secs(60), false);
|
||
|
||
let ts = timestamp::Timestamp::now();
|
||
|
||
// With duplicate detection disabled, should always pass
|
||
assert!(validator.check_duplicate("EVT1", ts).is_ok());
|
||
assert!(validator.check_duplicate("EVT1", ts).is_ok());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_symbol_edge_cases() {
|
||
let validator = validation::DataValidator::new(10.0, Duration::from_secs(60), false);
|
||
|
||
// Boundary length cases
|
||
assert!(validator.validate_symbol("A").is_ok()); // 1 char
|
||
assert!(validator.validate_symbol("ABCDEFGHIJK12").is_err()); // 13 chars
|
||
assert!(validator.validate_symbol("ABCDEFGHIJ12").is_ok()); // 12 chars
|
||
|
||
// Special valid characters
|
||
assert!(validator.validate_symbol("BRK.A").is_ok());
|
||
assert!(validator.validate_symbol("BTC-USD").is_ok());
|
||
assert!(validator.validate_symbol("STOCK123").is_ok());
|
||
|
||
// Invalid characters
|
||
assert!(validator.validate_symbol("BTC/USD").is_err());
|
||
assert!(validator.validate_symbol("STOCK@").is_err());
|
||
assert!(validator.validate_symbol("TEST#").is_err());
|
||
assert!(validator.validate_symbol("ABC_DEF").is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_timestamp_future() {
|
||
let validator = validation::DataValidator::new(10.0, Duration::from_secs(60), false);
|
||
|
||
// Future timestamp should pass (only checks for being too old)
|
||
let future_ts = timestamp::Timestamp {
|
||
nanos: timestamp::Timestamp::now().nanos + 60_000_000_000, // 60s in future
|
||
};
|
||
assert!(validator.validate_timestamp(future_ts).is_ok());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_price_zero_division() {
|
||
let validator = validation::DataValidator::new(10.0, Duration::from_secs(60), false);
|
||
|
||
// Division by zero protection
|
||
assert!(validator.validate_price_change(0.0, 100.0).is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_duplicate_ordering() {
|
||
let mut validator = validation::DataValidator::new(10.0, Duration::from_secs(60), true);
|
||
|
||
let ts1 = timestamp::Timestamp { nanos: 1000 };
|
||
let ts2 = timestamp::Timestamp { nanos: 2000 };
|
||
let ts3 = timestamp::Timestamp { nanos: 1500 };
|
||
|
||
assert!(validator.check_duplicate("EVT1", ts1).is_ok());
|
||
assert!(validator.check_duplicate("EVT1", ts2).is_ok());
|
||
|
||
// Out of order should fail
|
||
assert!(validator.check_duplicate("EVT1", ts3).is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_multiple_events() {
|
||
let mut validator = validation::DataValidator::new(10.0, Duration::from_secs(60), true);
|
||
|
||
let ts1 = timestamp::Timestamp { nanos: 1000 };
|
||
let ts2 = timestamp::Timestamp { nanos: 2000 };
|
||
|
||
// Different events should be independent
|
||
assert!(validator.check_duplicate("EVT1", ts1).is_ok());
|
||
assert!(validator.check_duplicate("EVT2", ts1).is_ok());
|
||
assert!(validator.check_duplicate("EVT1", ts2).is_ok());
|
||
assert!(validator.check_duplicate("EVT2", ts2).is_ok());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_symbol_unicode() {
|
||
let validator = validation::DataValidator::new(10.0, Duration::from_secs(60), false);
|
||
|
||
// Unicode characters should fail (only ASCII allowed)
|
||
assert!(validator.validate_symbol("СТОКЪ").is_err());
|
||
assert!(validator.validate_symbol("株式").is_err());
|
||
}
|
||
|
||
#[test]
|
||
fn test_validator_constructor_edge_cases() {
|
||
// Zero tolerance
|
||
let validator = validation::DataValidator::new(0.0, Duration::from_secs(60), true);
|
||
assert!(validator.validate_price_change(100.0, 100.0).is_ok());
|
||
assert!(validator.validate_price_change(100.0, 100.1).is_err());
|
||
|
||
// Zero timeout tolerance
|
||
let validator = validation::DataValidator::new(10.0, Duration::from_nanos(0), false);
|
||
let old_ts = timestamp::Timestamp {
|
||
nanos: timestamp::Timestamp::now().nanos.saturating_sub(1),
|
||
};
|
||
assert!(validator.validate_timestamp(old_ts).is_err());
|
||
}
|
||
|
||
// MONITORING TESTS (12 new tests)
|
||
#[test]
|
||
fn test_histogram_statistics() {
|
||
use monitoring::Histogram;
|
||
|
||
let mut h = Histogram::new();
|
||
for v in &[1.0, 2.0, 3.0, 4.0] {
|
||
h.record(*v);
|
||
}
|
||
let stats = h.stats();
|
||
assert_eq!(stats.count, 4);
|
||
assert_eq!(stats.min, 1.0);
|
||
assert_eq!(stats.max, 4.0);
|
||
assert!((stats.mean - 2.5).abs() < 1e-10);
|
||
// Median of [1.0, 2.0, 3.0, 4.0] is (2.0 + 3.0) / 2 = 2.5
|
||
assert_eq!(stats.p50, 2.5);
|
||
// p95 with linear interpolation: index=2.85, so 3.0 + (4.0-3.0)*0.85 = 3.85
|
||
assert!((stats.p95 - 3.85).abs() < 1e-10);
|
||
// p99 with linear interpolation: index=2.97, so 3.0 + (4.0-3.0)*0.97 = 3.97
|
||
assert!((stats.p99 - 3.97).abs() < 1e-10);
|
||
}
|
||
|
||
#[test]
|
||
fn test_histogram_empty_stats_default() {
|
||
let h = monitoring::Histogram::new();
|
||
let stats = h.stats();
|
||
assert_eq!(stats, monitoring::HistogramStats::default());
|
||
assert_eq!(stats.count, 0);
|
||
assert_eq!(stats.min, 0.0);
|
||
assert_eq!(stats.max, 0.0);
|
||
}
|
||
|
||
#[test]
|
||
fn test_histogram_single_value() {
|
||
use monitoring::Histogram;
|
||
|
||
let mut h = Histogram::new();
|
||
h.record(42.0);
|
||
let stats = h.stats();
|
||
|
||
assert_eq!(stats.count, 1);
|
||
assert_eq!(stats.min, 42.0);
|
||
assert_eq!(stats.max, 42.0);
|
||
assert_eq!(stats.mean, 42.0);
|
||
assert_eq!(stats.p50, 42.0);
|
||
assert_eq!(stats.p95, 42.0);
|
||
assert_eq!(stats.p99, 42.0);
|
||
}
|
||
|
||
#[test]
|
||
fn test_histogram_percentile_edge_cases() {
|
||
use monitoring::Histogram;
|
||
|
||
let mut h = Histogram::new();
|
||
// Add many values to test percentile calculation
|
||
for i in 1..=100 {
|
||
h.record(i as f64);
|
||
}
|
||
|
||
let stats = h.stats();
|
||
assert_eq!(stats.count, 100);
|
||
assert_eq!(stats.min, 1.0);
|
||
assert_eq!(stats.max, 100.0);
|
||
assert!((stats.mean - 50.5).abs() < 0.1);
|
||
assert!((stats.p50 - 50.0).abs() < 1.0);
|
||
assert!((stats.p95 - 95.0).abs() < 1.0);
|
||
assert!((stats.p99 - 99.0).abs() < 1.0);
|
||
}
|
||
|
||
#[test]
|
||
fn test_metrics_collector_concurrent_access() {
|
||
use monitoring::MetricsCollector;
|
||
use std::sync::Arc;
|
||
use std::thread;
|
||
|
||
let metrics = Arc::new(MetricsCollector::new());
|
||
let mut handles: Vec<thread::JoinHandle<()>> = vec![];
|
||
|
||
// Spawn multiple threads incrementing counters
|
||
for i in 0..10 {
|
||
let metrics_clone = Arc::clone(&metrics);
|
||
let handle = thread::spawn(move || {
|
||
for _ in 0..100 {
|
||
metrics_clone.increment_counter("concurrent_counter", 1);
|
||
metrics_clone.set_gauge(&format!("gauge_{}", i), i);
|
||
metrics_clone.record_histogram("latency", i as f64 * 0.1);
|
||
}
|
||
});
|
||
handles.push(handle);
|
||
}
|
||
|
||
for handle in handles {
|
||
handle.join().unwrap();
|
||
}
|
||
|
||
assert_eq!(metrics.get_counter("concurrent_counter"), 1000);
|
||
let snapshot = metrics.export_metrics();
|
||
assert!(snapshot.counters.len() > 0);
|
||
assert!(snapshot.gauges.len() > 0);
|
||
assert!(snapshot.histograms.len() > 0);
|
||
}
|
||
|
||
#[test]
|
||
fn test_metrics_collector_nonexistent_metrics() {
|
||
let metrics = monitoring::MetricsCollector::new();
|
||
|
||
assert_eq!(metrics.get_counter("nonexistent"), 0);
|
||
assert_eq!(metrics.get_gauge("nonexistent"), 0);
|
||
assert!(metrics.get_histogram_stats("nonexistent").is_none());
|
||
}
|
||
|
||
#[test]
|
||
fn test_metrics_snapshot_serialization() {
|
||
use monitoring::MetricsCollector;
|
||
|
||
let metrics = MetricsCollector::new();
|
||
metrics.increment_counter("test_counter", 42);
|
||
metrics.set_gauge("test_gauge", 123);
|
||
metrics.record_histogram("test_histogram", 1.5);
|
||
|
||
let snapshot = metrics.export_metrics();
|
||
let json = serde_json::to_string(&snapshot).unwrap();
|
||
let deserialized: monitoring::MetricsSnapshot = serde_json::from_str(&json).unwrap();
|
||
|
||
assert_eq!(deserialized.counters.get("test_counter"), Some(&42));
|
||
assert_eq!(deserialized.gauges.get("test_gauge"), Some(&123));
|
||
assert!(deserialized.histograms.get("test_histogram").is_some());
|
||
}
|
||
|
||
#[test]
|
||
fn test_latency_measurer() {
|
||
use monitoring::{LatencyMeasurer, MetricsCollector};
|
||
|
||
let metrics = MetricsCollector::new();
|
||
let measurer = LatencyMeasurer::start("test_operation".to_string(), metrics.clone());
|
||
|
||
// Simulate some work
|
||
std::thread::sleep(Duration::from_millis(1));
|
||
|
||
let duration = measurer.finish();
|
||
assert!(duration.as_micros() > 0);
|
||
|
||
// Check that histogram was recorded
|
||
let stats = metrics.get_histogram_stats("test_operation_latency_us");
|
||
assert!(stats.is_some());
|
||
let stats = stats.unwrap();
|
||
assert_eq!(stats.count, 1);
|
||
assert!(stats.mean > 0.0);
|
||
}
|
||
|
||
#[test]
|
||
fn test_histogram_stats_display() {
|
||
use monitoring::HistogramStats;
|
||
|
||
let stats = HistogramStats {
|
||
count: 100,
|
||
min: 1.0,
|
||
max: 10.0,
|
||
mean: 5.5,
|
||
p50: 5.0,
|
||
p95: 9.5,
|
||
p99: 9.9,
|
||
};
|
||
|
||
let json = serde_json::to_string(&stats).unwrap();
|
||
let deserialized: HistogramStats = serde_json::from_str(&json).unwrap();
|
||
assert_eq!(stats.count, deserialized.count);
|
||
assert!((stats.mean - deserialized.mean).abs() < f64::EPSILON);
|
||
}
|
||
|
||
#[test]
|
||
fn test_metrics_collector_large_values() {
|
||
let metrics = monitoring::MetricsCollector::new();
|
||
|
||
// Test with large values
|
||
metrics.increment_counter("large_counter", u64::MAX / 2);
|
||
metrics.increment_counter("large_counter", u64::MAX / 2);
|
||
|
||
// Should wrap around
|
||
let value = metrics.get_counter("large_counter");
|
||
assert_eq!(value, u64::MAX.wrapping_sub(1));
|
||
|
||
// Test large gauge
|
||
metrics.set_gauge("large_gauge", u64::MAX);
|
||
assert_eq!(metrics.get_gauge("large_gauge"), u64::MAX);
|
||
}
|
||
|
||
#[test]
|
||
fn test_histogram_extreme_values() {
|
||
use monitoring::Histogram;
|
||
|
||
let mut h = Histogram::new();
|
||
h.record(f64::MIN);
|
||
h.record(f64::MAX);
|
||
h.record(0.0);
|
||
|
||
let stats = h.stats();
|
||
assert_eq!(stats.count, 3);
|
||
assert_eq!(stats.min, f64::MIN);
|
||
assert_eq!(stats.max, f64::MAX);
|
||
}
|
||
|
||
// LOCKFREE QUEUE TESTS (8 new tests)
|
||
#[test]
|
||
fn test_lockfree_queue_overflow() {
|
||
let q = lockfree::LockFreeQueue::new(2);
|
||
|
||
assert!(q.push(1));
|
||
assert!(q.push(2));
|
||
// third push should fail
|
||
assert!(!q.push(3));
|
||
assert!(q.has_overflowed());
|
||
|
||
// Pop remaining elements
|
||
assert_eq!(q.pop(), Some(1));
|
||
assert_eq!(q.pop(), Some(2));
|
||
assert!(q.is_empty());
|
||
q.reset_overflow();
|
||
assert!(!q.has_overflowed());
|
||
}
|
||
|
||
#[test]
|
||
fn test_lockfree_queue_concurrent_push_pop() {
|
||
use std::sync::Arc;
|
||
use std::thread;
|
||
|
||
let queue = Arc::new(lockfree::LockFreeQueue::new(1000));
|
||
let mut handles: Vec<thread::JoinHandle<()>> = vec![];
|
||
|
||
// Producer threads
|
||
for i in 0..5 {
|
||
let queue_clone = Arc::clone(&queue);
|
||
let handle = thread::spawn(move || {
|
||
for j in 0..100 {
|
||
let value = i * 100 + j;
|
||
while !queue_clone.push(value) {
|
||
thread::yield_now();
|
||
}
|
||
}
|
||
});
|
||
handles.push(handle);
|
||
}
|
||
|
||
// Consumer thread
|
||
let queue_clone = Arc::clone(&queue);
|
||
let consumer_handle = thread::spawn(move || {
|
||
let mut consumed = 0;
|
||
while consumed < 500 {
|
||
if queue_clone.pop().is_some() {
|
||
consumed += 1;
|
||
}
|
||
thread::yield_now();
|
||
}
|
||
consumed
|
||
});
|
||
|
||
for handle in handles {
|
||
handle.join().unwrap();
|
||
}
|
||
|
||
let consumed = consumer_handle.join().unwrap();
|
||
assert_eq!(consumed, 500);
|
||
}
|
||
|
||
#[test]
|
||
fn test_lockfree_queue_size_consistency() {
|
||
let queue = lockfree::LockFreeQueue::new(10);
|
||
|
||
assert_eq!(queue.len(), 0);
|
||
assert!(queue.is_empty());
|
||
|
||
queue.push("item1");
|
||
assert_eq!(queue.len(), 1);
|
||
assert!(!queue.is_empty());
|
||
|
||
queue.push("item2");
|
||
assert_eq!(queue.len(), 2);
|
||
|
||
queue.pop();
|
||
assert_eq!(queue.len(), 1);
|
||
|
||
queue.pop();
|
||
assert_eq!(queue.len(), 0);
|
||
assert!(queue.is_empty());
|
||
}
|
||
|
||
#[test]
|
||
fn test_lockfree_queue_fifo_order() {
|
||
let queue = lockfree::LockFreeQueue::new(5);
|
||
|
||
for i in 1..=5 {
|
||
assert!(queue.push(i));
|
||
}
|
||
|
||
for i in 1..=5 {
|
||
assert_eq!(queue.pop(), Some(i));
|
||
}
|
||
|
||
assert_eq!(queue.pop(), None);
|
||
}
|
||
|
||
#[test]
|
||
fn test_lockfree_queue_empty_pop() {
|
||
let queue = lockfree::LockFreeQueue::<i32>::new(10);
|
||
assert_eq!(queue.pop(), None);
|
||
assert!(queue.is_empty());
|
||
}
|
||
|
||
#[test]
|
||
fn test_lockfree_queue_max_size_one() {
|
||
let queue = lockfree::LockFreeQueue::new(1);
|
||
|
||
assert!(queue.push(42));
|
||
assert!(!queue.push(43));
|
||
assert!(queue.has_overflowed());
|
||
|
||
assert_eq!(queue.pop(), Some(42));
|
||
assert!(queue.is_empty());
|
||
|
||
// After reset, should work again
|
||
queue.reset_overflow();
|
||
assert!(!queue.has_overflowed());
|
||
assert!(queue.push(44));
|
||
}
|
||
|
||
#[test]
|
||
fn test_lockfree_queue_zero_size() {
|
||
let queue = lockfree::LockFreeQueue::<i32>::new(0);
|
||
|
||
assert!(!queue.push(1));
|
||
assert!(queue.has_overflowed());
|
||
assert_eq!(queue.pop(), None);
|
||
}
|
||
|
||
#[test]
|
||
fn test_lockfree_queue_stress_test() {
|
||
use std::sync::Arc;
|
||
use std::thread;
|
||
|
||
let queue = Arc::new(lockfree::LockFreeQueue::new(100));
|
||
let iterations = 1000;
|
||
let mut _handles: Vec<thread::JoinHandle<()>> = vec![];
|
||
|
||
// Single producer, single consumer stress test
|
||
let producer_queue = Arc::clone(&queue);
|
||
let producer = thread::spawn(move || {
|
||
for i in 0..iterations {
|
||
while !producer_queue.push(i) {
|
||
thread::yield_now();
|
||
}
|
||
}
|
||
});
|
||
|
||
let consumer_queue = Arc::clone(&queue);
|
||
let consumer = thread::spawn(move || {
|
||
let mut received = Vec::new();
|
||
while received.len() < iterations {
|
||
if let Some(value) = consumer_queue.pop() {
|
||
received.push(value);
|
||
}
|
||
thread::yield_now();
|
||
}
|
||
received
|
||
});
|
||
|
||
producer.join().unwrap();
|
||
let received = consumer.join().unwrap();
|
||
|
||
assert_eq!(received.len(), iterations);
|
||
// Verify FIFO order
|
||
for (i, value) in received.into_iter().enumerate() {
|
||
assert_eq!(value, i);
|
||
}
|
||
}
|
||
|
||
// NETWORK TESTS (8 new tests)
|
||
#[tokio::test]
|
||
async fn test_connection_helper_timeout() {
|
||
use network::ConnectionHelper;
|
||
|
||
tokio::time::pause();
|
||
|
||
let helper = ConnectionHelper::default();
|
||
let timeout_task = helper.connect_with_timeout(
|
||
|| async { std::future::pending::<std::result::Result<(), std::io::Error>>().await },
|
||
Duration::from_millis(50),
|
||
);
|
||
|
||
// Advance time past timeout
|
||
tokio::time::advance(Duration::from_millis(51)).await;
|
||
|
||
let err = timeout_task.await;
|
||
assert!(err.is_err(), "expected timeout error");
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_connection_helper_successful_connection() {
|
||
use network::ConnectionHelper;
|
||
|
||
let helper = ConnectionHelper::default();
|
||
let result = helper
|
||
.connect_with_timeout(
|
||
|| async { Ok::<&str, std::io::Error>("Connected") },
|
||
Duration::from_millis(100),
|
||
)
|
||
.await;
|
||
assert_eq!(result.unwrap(), "Connected");
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_connection_helper_retry_exhausted() {
|
||
use network::ConnectionHelper;
|
||
|
||
tokio::time::pause();
|
||
|
||
let helper = ConnectionHelper::new(
|
||
3, // max_attempts
|
||
Duration::from_millis(10), // initial_delay
|
||
Duration::from_millis(100), // max_delay
|
||
2.0, // backoff_multiplier
|
||
0.0, // no jitter for deterministic test
|
||
);
|
||
|
||
let mut attempts = 0;
|
||
let result = helper
|
||
.retry_connect(|| {
|
||
attempts += 1;
|
||
async move { Err::<(), &str>("Always fails") }
|
||
})
|
||
.await;
|
||
|
||
assert!(result.is_err());
|
||
assert_eq!(attempts, 3);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_connection_helper_eventual_success() {
|
||
use network::ConnectionHelper;
|
||
|
||
tokio::time::pause();
|
||
|
||
let helper = ConnectionHelper::new(
|
||
5, // max_attempts
|
||
Duration::from_millis(1), // initial_delay
|
||
Duration::from_millis(10), // max_delay
|
||
1.5, // backoff_multiplier
|
||
0.0, // no jitter for predictable timing
|
||
);
|
||
|
||
let mut attempts = 0;
|
||
|
||
let result = helper
|
||
.retry_connect(|| {
|
||
attempts += 1;
|
||
async move {
|
||
if attempts < 3 {
|
||
Err("Not yet")
|
||
} else {
|
||
Ok("Finally connected")
|
||
}
|
||
}
|
||
})
|
||
.await;
|
||
|
||
assert_eq!(result.unwrap(), "Finally connected");
|
||
assert_eq!(attempts, 3);
|
||
// With paused time, we don't need to check elapsed time
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_connection_helper_backoff_progression() {
|
||
use network::ConnectionHelper;
|
||
|
||
tokio::time::pause();
|
||
|
||
let helper = ConnectionHelper::new(
|
||
4, // max_attempts
|
||
Duration::from_millis(10), // initial_delay
|
||
Duration::from_millis(100), // max_delay
|
||
2.0, // backoff_multiplier
|
||
0.0, // no jitter
|
||
);
|
||
|
||
let mut attempts = 0;
|
||
|
||
let result = helper
|
||
.retry_connect(|| {
|
||
attempts += 1;
|
||
async move { Err::<(), &str>("Keep failing") }
|
||
})
|
||
.await;
|
||
|
||
assert!(result.is_err());
|
||
assert_eq!(attempts, 4);
|
||
// With paused time, we verify logical behavior rather than timing
|
||
}
|
||
|
||
#[test]
|
||
fn test_connection_helper_default() {
|
||
let helper1 = network::ConnectionHelper::default();
|
||
let helper2 = network::ConnectionHelper::new(
|
||
10,
|
||
Duration::from_millis(1000),
|
||
Duration::from_millis(60000),
|
||
2.0,
|
||
0.1,
|
||
);
|
||
|
||
// Can't directly compare structs, but we can test they have same behavior
|
||
// by checking they both exist and are constructible
|
||
let _ = helper1;
|
||
let _ = helper2;
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_connection_helper_zero_attempts() {
|
||
use network::ConnectionHelper;
|
||
|
||
tokio::time::pause();
|
||
|
||
let helper = ConnectionHelper::new(
|
||
0, // max_attempts (edge case)
|
||
Duration::from_millis(10),
|
||
Duration::from_millis(100),
|
||
2.0,
|
||
0.0,
|
||
);
|
||
|
||
let mut attempts = 0;
|
||
let result = helper
|
||
.retry_connect(|| {
|
||
attempts += 1;
|
||
async move { Err::<(), &str>("Attempted despite 0 max") }
|
||
})
|
||
.await;
|
||
|
||
assert!(result.is_err());
|
||
// Current implementation attempts at least once even with 0 max_attempts
|
||
// This is actually reasonable behavior (try at least once)
|
||
assert_eq!(attempts, 1);
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_connection_helper_jitter() {
|
||
use network::ConnectionHelper;
|
||
|
||
tokio::time::pause();
|
||
|
||
let helper = ConnectionHelper::new(
|
||
3,
|
||
Duration::from_millis(10),
|
||
Duration::from_millis(100),
|
||
1.0, // no exponential backoff
|
||
0.0, // no jitter for deterministic test
|
||
);
|
||
|
||
let mut attempts = 0;
|
||
|
||
let result = helper
|
||
.retry_connect(|| {
|
||
attempts += 1;
|
||
async move { Err::<(), &str>("Keep failing") }
|
||
})
|
||
.await;
|
||
|
||
assert!(result.is_err());
|
||
assert_eq!(attempts, 3);
|
||
// With paused time, we verify logical behavior rather than jitter timing
|
||
}
|
||
}
|