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

fix(trading_service): hold write lock across position update to prevent fill race

Browse Source 
The update_position method previously acquired a read lock to get the
Arc<AtomicPosition>, dropped it, then called update_with_execution
outside any lock. Two concurrent fills for the same position could both
load the same old_quantity via Acquire, compute independent new quantities,
and the last Release store would silently discard the other fill — causing
the position to show e.g. 10 shares when it should show 20.

Now holds the positions write lock across the entire get-or-create +
update_with_execution sequence, serializing concurrent fills per position.
The validate_position_update call remains outside the lock to minimize
hold time (it does its own async reads and risk checks).

Also removes excessive per-step RDTSC latency tracking from the critical
path (total latency tracking is preserved) and adds a multi-threaded
regression test that spawns 10 concurrent +1 fills and asserts the final
quantity equals 10.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-23 20:56:49 +01:00
parent 2f83e57826
commit 330348e84d
1 changed files with 106 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
services/trading_service/src/core/position_manager.rs
View File

@@ -328,6 +328,12 @@ impl PositionManager {
}
/// Update position with trade execution - REAL PRODUCTION IMPLEMENTATION
///
/// SAFETY: Holds write lock across the entire get-or-create + update_with_execution
/// sequence to prevent concurrent fills from racing on the same position. Without
/// this, two fills arriving simultaneously could both read the same old quantity,
/// compute independent new quantities, and the last store wins — silently dropping
/// one fill's effect (e.g., position shows 10 shares when it should be 20).
pub async fn update_position(
&self,
account_id: &str,
@@ -335,87 +341,52 @@ impl PositionManager {
quantity_delta: i64,
execution_price: f64,
) -> Result<PositionUpdate, PositionError> {
// CRITICAL PATH: <14ns RDTSC timing for position updates
let update_start = HardwareTimestamp::now();
let mut latency_tracker = LatencyMeasurement::start();
// Key generation - RDTSC timed (target: <2ns)
let key_start = HardwareTimestamp::now();
let position_key = format!("{}:{}", account_id, symbol);
let timestamp_ns = HardwareTimestamp::now().as_nanos();
let sequence = self.sequence_generator.next();
let key_latency = HardwareTimestamp::now().latency_ns(&key_start);
if key_latency > 2 {
warn!(
"Position key generation exceeded 2ns target: {}ns",
key_latency
);
}
// REAL RISK CHECK - Position limits and exposure validation - RDTSC timed (target: <6ns)
let risk_start = HardwareTimestamp::now();
// Validate BEFORE acquiring write lock to minimize lock hold time.
// validate_position_update does its own async reads and risk checks.
self.validate_position_update(account_id, symbol, quantity_delta, execution_price)
.await?;
let risk_latency = HardwareTimestamp::now().latency_ns(&risk_start);
if risk_latency > 6 {
warn!("Risk validation exceeded 6ns target: {}ns", risk_latency);
}
// Hold write lock for the entire get-or-create + update_with_execution sequence.
// This serializes concurrent fills for the SAME position, preventing the race
// where two fills load the same old_quantity and the last store wins.
let mut positions = self.positions.write().await;
// Get or create position - RDTSC timed (target: <4ns)
let lookup_start = HardwareTimestamp::now();
let position = {
let positions = self.positions.read().await;
if let Some(pos) = positions.get(&position_key) {
Arc::clone(pos)
} else {
drop(positions);
// Create new position
let symbol_hash = self.get_symbol_hash(symbol).await;
let account_hash = self.get_account_hash(account_id).await;
let new_position = Arc::new(AtomicPosition::new(symbol_hash, account_hash));
{
let mut positions = self.positions.write().await;
positions.insert(position_key.clone(), Arc::clone(&new_position));
self.position_count.fetch_add(1, Ordering::Relaxed);
}
new_position
}
let position = if let Some(pos) = positions.get(&position_key) {
Arc::clone(pos)
} else {
// Create new position (hash lookups are cheap cached ops)
let symbol_hash = self.get_symbol_hash(symbol).await;
let account_hash = self.get_account_hash(account_id).await;
let new_position = Arc::new(AtomicPosition::new(symbol_hash, account_hash));
positions.insert(position_key.clone(), Arc::clone(&new_position));
self.position_count.fetch_add(1, Ordering::Relaxed);
new_position
};
let lookup_latency = HardwareTimestamp::now().latency_ns(&lookup_start);
if lookup_latency > 4 {
warn!("Position lookup exceeded 4ns target: {}ns", lookup_latency);
}
// Perform atomic update - RDTSC timed (target: <3ns)
let atomic_start = HardwareTimestamp::now();
// Still under write lock — no concurrent fill can interleave
let update_result = position.update_with_execution(
quantity_delta,
execution_price,
timestamp_ns,
sequence,
)?;
let atomic_latency = HardwareTimestamp::now().latency_ns(&atomic_start);
if atomic_latency > 3 {
warn!(
"Atomic position update exceeded 3ns target: {}ns",
atomic_latency
);
}
// Explicit drop after critical section — post-update work runs unlocked
drop(positions);
// REAL PERFORMANCE TRACKING with comprehensive RDTSC timing
// Performance tracking
let total_update_latency = HardwareTimestamp::now().latency_ns(&update_start);
self.update_count.fetch_add(1, Ordering::Relaxed);
let elapsed_ns = latency_tracker.finish();
self.metrics.record_operation_time(elapsed_ns);
// CRITICAL: Track total position update latency (target: <14ns)
if total_update_latency > 14 {
error!(
"Position update EXCEEDED 14ns target: {}ns for {}",
@@ -428,24 +399,17 @@ impl PositionManager {
);
}
// REAL COMPLIANCE LOGGING with detailed timing
info!("Position updated: {} delta={} price={} avg_price={} pnl_delta={} in {}ns (total: {}ns, atomic: {}ns)",
position_key, quantity_delta, execution_price,
update_result.new_avg_price, update_result.realized_pnl_delta,
elapsed_ns, total_update_latency, atomic_latency);
// Compliance logging
info!(
"Position updated: {} delta={} price={} avg_price={} pnl_delta={} in {}ns (total: {}ns)",
position_key, quantity_delta, execution_price,
update_result.new_avg_price, update_result.realized_pnl_delta,
elapsed_ns, total_update_latency
);
// REAL RISK MONITORING - Check position concentration (RDTSC timed)
let risk_monitor_start = HardwareTimestamp::now();
// Post-update risk monitoring (outside the lock)
self.monitor_position_risk(account_id, symbol, &update_result)
.await?;
let risk_monitor_latency = HardwareTimestamp::now().latency_ns(&risk_monitor_start);
if risk_monitor_latency > 5 {
warn!(
"Risk monitoring exceeded 5ns target: {}ns",
risk_monitor_latency
);
}
Ok(update_result)
}
@@ -1116,4 +1080,76 @@ mod tests {
assert_eq!(snapshot.quantity, 100);
assert_eq!(snapshot.market_price, 51000.0);
}
/// Regression test for C11: concurrent fills must not lose updates.
///
/// Before the fix, `update_position` acquired a read lock to get the
/// Arc<AtomicPosition>, dropped it, then called `update_with_execution`
/// outside any lock. Two concurrent fills could both load the same
/// old_quantity, compute independent new_quantity values, and the last
/// store would silently discard the other fill.
///
/// The fix holds the write lock across the entire get-or-create +
/// update_with_execution sequence, serializing concurrent fills.
///
/// NOTE: With tokio's single-threaded test runtime, tasks yield at
/// .await points but don't truly run in parallel. The multi_thread
/// flavor is required to expose real interleaving. We use
/// `#[tokio::test(flavor = "multi_thread", worker_threads = 4)]` to
/// maximize the chance of triggering the race.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_concurrent_fills_no_lost_updates() {
let config = TradingConfig::default();
let config_manager = Arc::new(config::manager::ConfigManager::new(
config::manager::ServiceConfig {
name: "test".to_string(),
environment: "test".to_string(),
version: "1.0.0".to_string(),
settings: serde_json::json!({}),
},
));
let manager = Arc::new(
PositionManager::new(config, config_manager)
.await
.expect("PositionManager should be created for concurrency test"),
);
let num_fills: i64 = 10;
let fill_quantity: i64 = 1;
let fill_price = 100.0;
// Spawn num_fills concurrent tasks, each adding +1 to the same position
let mut handles = Vec::with_capacity(num_fills as usize);
for _ in 0..num_fills {
let mgr = Arc::clone(&manager);
handles.push(tokio::spawn(async move {
mgr.update_position("concurrent-acct", "TESTSYM", fill_quantity, fill_price)
.await
}));
}
// Collect results — all should succeed
for handle in handles {
let result = handle
.await
.expect("Task should not panic");
assert!(result.is_ok(), "Each fill should succeed");
}
// The final position quantity must equal num_fills * fill_quantity.
// Before the fix, under real parallelism this could be less than 10
// because two fills would race on load+store of the quantity.
let snapshot = manager
.get_position("concurrent-acct", "TESTSYM")
.await
.expect("Position should exist after concurrent fills");
assert_eq!(
snapshot.quantity,
num_fills * fill_quantity,
"All {} fills must be reflected in the final position quantity (got {})",
num_fills,
snapshot.quantity,
);
}
}