From 330348e84d7de00177058d0484a102e76741819a Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 23 Feb 2026 20:56:49 +0100 Subject: [PATCH] fix(trading_service): hold write lock across position update to prevent fill race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The update_position method previously acquired a read lock to get the Arc, 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 --- .../src/core/position_manager.rs | 176 +++++++++++------- 1 file changed, 106 insertions(+), 70 deletions(-) diff --git a/services/trading_service/src/core/position_manager.rs b/services/trading_service/src/core/position_manager.rs index 0aba55309..b35cba726 100644 --- a/services/trading_service/src/core/position_manager.rs +++ b/services/trading_service/src/core/position_manager.rs @@ -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 { - // 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, 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, + ); + } }