From 248b070c9e8b2cd0159e5f5f42e926602b8b55f8 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 13:54:26 +0200 Subject: [PATCH 1/4] fix(price_level): make execution-statistics failures non-partial and observable record_execution validated the multiplication and timestamps first but then mutated several independent atomics with fallible checked additions: a later counter overflow left the earlier counters already advanced (orders_executed and quantity_executed could advance while value_executed rejected), and match_order discarded the returned error, so a trade succeeded with silently inconsistent statistics. The additive aggregates are now committed with fetch_update(checked_add) CAS loops and, on any later failure, the already-committed prefix is rolled back with fetch_sub of exactly this call's contribution (the crate's #111/#113 call-backed rollback argument -- documented as contingent on reset()'s new quiescence contract). last_execution_time is stored only after every additive commit succeeds. An accepted execution therefore contributes to ALL aggregates or NONE in the committed state; the transient prefix window is documented honestly. match_order no longer discards the error: it logs at WARN (the match is not aborted -- the trade is already committed and is never unwound) and the new sticky stats_degraded flag records that an execution's statistics were dropped. The flag is queryable, carried through Clone / Display / FromStr / serde, and is serialized ONLY when true so a pre-#117 v2 snapshot package re-serializes byte-identically and its SHA-256 still validates -- old snapshots restore unchanged (guarded by an old-format compatibility test). Tests: forced overflow per aggregate (all-or-nothing, flag set, Err), Barrier-controlled concurrent boundary (exactly K symmetric-headroom successes, quantity/value lockstep), match_order integration (trade intact, flag observable and round-tripped), serde omit-when-false guards, torn-read caveats on the average accessors. Closes #117 --- src/price_level/level.rs | 37 ++++- src/price_level/statistics.rs | 213 +++++++++++++++++++++++----- src/price_level/tests/level.rs | 103 ++++++++++++++ src/price_level/tests/statistics.rs | 177 +++++++++++++++++++++++ 4 files changed, 494 insertions(+), 36 deletions(-) diff --git a/src/price_level/level.rs b/src/price_level/level.rs index dc9a84a..00e550d 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -957,6 +957,18 @@ impl PriceLevel { /// admissions arrive from one logical path (see the type-level note on /// [`PriceLevel`]), because the side is derived from the live queue rather /// than stored. + /// + /// # Statistics + /// + /// Per-level execution statistics are recorded **all-or-nothing** and can + /// never fail the match: the trade is already committed when + /// `PriceLevelStatistics::record_execution` runs. If recording overflows a + /// statistics counter, that execution's contribution is dropped atomically + /// (it advances every aggregate or none), the drop is logged at `WARN` (a + /// recoverable anomaly, not an aborted match), and the level's + /// `PriceLevelStatistics::stats_degraded` flag is set (sticky, + /// snapshot-persisted) so the under-count is observable. The emitted trades + /// and the `MatchResult` are unaffected (issue #117). pub fn match_order( &self, incoming_quantity: u64, @@ -1365,12 +1377,33 @@ impl PriceLevel { result.add_filled_order_id(data.maker_id); } - let _ = self.stats.record_execution( + // The trade is already committed (added to `result` and + // the queue mutated) — statistics recording cannot fail + // it retroactively. Recording is all-or-nothing (issue + // #117): on overflow the execution's contribution is + // dropped atomically and a sticky `stats_degraded` flag + // is set on the level's statistics (observable via + // `PriceLevelStatistics::stats_degraded`), leaving the + // trade stream unaffected. Log the drop rather than + // discard it silently. + if let Err(err) = self.stats.record_execution( data.consumed, data.maker_price, data.maker_timestamp, timestamp.as_u64(), - ); + ) { + // WARN, not ERROR: the match is not aborted — this is + // a recoverable observability anomaly flagged by the + // sticky degraded flag (the trade is committed). + tracing::warn!( + price = self.price, + taker_order_id = %taker_order_id, + maker_order_id = %data.maker_id, + consumed = data.consumed, + error = %err, + "execution statistics dropped (all-or-nothing); level stats marked degraded — trade unaffected" + ); + } } remaining = new_remaining; diff --git a/src/price_level/statistics.rs b/src/price_level/statistics.rs index f635b47..65d75fc 100644 --- a/src/price_level/statistics.rs +++ b/src/price_level/statistics.rs @@ -4,7 +4,7 @@ use serde::ser::SerializeStruct; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use std::fmt; use std::str::FromStr; -use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; /// Tracks performance statistics for a price level. @@ -60,6 +60,15 @@ pub struct PriceLevelStatistics { /// Sum of waiting times for orders sum_waiting_time: AtomicU64, + + /// Sticky flag: set once and never cleared (except by [`reset`](Self::reset)) + /// when an execution's statistics contribution was **dropped** — a + /// [`record_execution`](Self::record_execution) that failed validation or + /// overflowed a counter and so contributed to NONE of the aggregates + /// (all-or-nothing, issue #117). The trade itself is unaffected; this flag + /// is the observable signal that the recorded aggregates under-count the + /// true executions. Serialized so it round-trips through a snapshot. + stats_degraded: AtomicBool, } impl PriceLevelStatistics { @@ -124,6 +133,7 @@ impl PriceLevelStatistics { last_execution_time: AtomicU64::new(0), first_arrival_time: AtomicU64::new(current_time), sum_waiting_time: AtomicU64::new(0), + stats_degraded: AtomicBool::new(false), } } @@ -147,6 +157,26 @@ impl PriceLevelStatistics { /// /// [`Trade`]: crate::execution::Trade /// + /// # All-or-nothing (issue #117) + /// + /// An accepted execution contributes to **every** aggregate, or to **none**. + /// If a later counter overflows after earlier ones already advanced, this + /// rolls the committed prefix back (a `fetch_sub` of exactly what this call + /// added — sound by the same call-backed reservation argument as the #111 / + /// #113 rollbacks: the subtracted units are exactly the units this call + /// added, so the undo is commutative with concurrent `record_execution` + /// deltas). That call-backed argument assumes no concurrent + /// [`reset`](Self::reset) `store(0)` races the rollback — see the quiescence + /// contract on `reset`. So a caller never observes a partial contribution in + /// the final state. On any failure — a validation error or a counter overflow — + /// the sticky [`stats_degraded`](Self::stats_degraded) flag is set: the + /// dropped execution is then observable, even though the caller + /// (`PriceLevel::match_order`) cannot fail the already-committed trade. + /// Because these are independent lock-free atomics, a *concurrent* reader may + /// still glimpse a prefix transiently before its rollback (the same window + /// the #111 reserve-then-rollback admission has); the guarantee is on the + /// committed final state, not on the transient. + /// /// # Errors /// /// Returns [`PriceLevelError::InvalidOperation`] if any of the counter @@ -163,44 +193,81 @@ impl PriceLevelStatistics { let current_time = execution_timestamp; // Validate everything that can fail BEFORE mutating any counter, so a - // rejected record leaves the statistics untouched. `match_order` ignores - // the returned `Result`, so any partial side effect would silently - // corrupt the stats. + // rejected record leaves the statistics untouched. Any failure marks the + // stats degraded: this execution's contribution is being dropped. let waiting_time = if order_timestamp > 0 { - Some(current_time.checked_sub(order_timestamp).ok_or_else(|| { - PriceLevelError::InvalidOperation { - message: format!( - "order timestamp {} is in the future of current time {}", - order_timestamp, current_time - ), + match current_time.checked_sub(order_timestamp) { + Some(value) => Some(value), + None => { + self.stats_degraded.store(true, Ordering::Relaxed); + return Err(PriceLevelError::InvalidOperation { + message: format!( + "order timestamp {order_timestamp} is in the future of current time {current_time}" + ), + }); } - })?) + } } else { None }; - let value_u128 = u128::from(quantity).checked_mul(price).ok_or_else(|| { - PriceLevelError::InvalidOperation { - message: "value_executed multiplication overflow".to_string(), + let value_u64 = match u128::from(quantity) + .checked_mul(price) + .and_then(|value| u64::try_from(value).ok()) + { + Some(value) => value, + None => { + self.stats_degraded.store(true, Ordering::Relaxed); + return Err(PriceLevelError::InvalidOperation { + message: "value_executed overflow (quantity * price exceeds u64 storage)" + .to_string(), + }); } - })?; - - let value_u64 = - u64::try_from(value_u128).map_err(|_| PriceLevelError::InvalidOperation { - message: "value_executed exceeds u64 storage".to_string(), - })?; + }; - // All fallible validation passed — now apply the mutations. (The only - // residual failure mode is a counter nearing `u64::MAX`, inherent to - // independent lock-free atomics.) + // Commit the additive aggregates with a rollback of the already-committed + // prefix on a later overflow (all-or-nothing). `orders_executed` is a + // `usize` counter that cannot overflow in practice, but it is part of the + // transaction so it is rolled back too. `last_execution_time` is a + // non-additive "latest" store, applied only after every additive commit + // succeeds so a rejected record never advances it. self.orders_executed.fetch_add(1, Ordering::Relaxed); - Self::checked_fetch_add_u64(&self.quantity_executed, quantity, "quantity_executed")?; - Self::checked_fetch_add_u64(&self.value_executed, value_u64, "value_executed")?; + + if let Err(err) = + Self::checked_fetch_add_u64(&self.quantity_executed, quantity, "quantity_executed") + { + self.orders_executed.fetch_sub(1, Ordering::Relaxed); + self.stats_degraded.store(true, Ordering::Relaxed); + return Err(err); + } + + if let Err(err) = + Self::checked_fetch_add_u64(&self.value_executed, value_u64, "value_executed") + { + self.quantity_executed + .fetch_sub(quantity, Ordering::Relaxed); + self.orders_executed.fetch_sub(1, Ordering::Relaxed); + self.stats_degraded.store(true, Ordering::Relaxed); + return Err(err); + } + + if let Some(waiting_time) = waiting_time + && let Err(err) = Self::checked_fetch_add_u64( + &self.sum_waiting_time, + waiting_time, + "sum_waiting_time", + ) + { + self.value_executed.fetch_sub(value_u64, Ordering::Relaxed); + self.quantity_executed + .fetch_sub(quantity, Ordering::Relaxed); + self.orders_executed.fetch_sub(1, Ordering::Relaxed); + self.stats_degraded.store(true, Ordering::Relaxed); + return Err(err); + } + self.last_execution_time .store(current_time, Ordering::Relaxed); - if let Some(waiting_time) = waiting_time { - Self::checked_fetch_add_u64(&self.sum_waiting_time, waiting_time, "sum_waiting_time")?; - } Ok(()) } @@ -268,7 +335,25 @@ impl PriceLevelStatistics { self.sum_waiting_time.load(Ordering::Relaxed) } - /// Get average execution price + /// Returns `true` if the recorded statistics are **degraded** — at least one + /// execution's contribution was dropped all-or-nothing (a validation error + /// or a counter overflow in [`record_execution`](Self::record_execution), + /// issue #117). + /// + /// The flag is sticky: once set it stays set until [`reset`](Self::reset). + /// When `true`, the aggregate counters under-count the true executions; the + /// emitted trade stream is unaffected. Cleared by `reset`. + #[must_use] + pub fn stats_degraded(&self) -> bool { + self.stats_degraded.load(Ordering::Relaxed) + } + + /// Get average execution price. + /// + /// Reads `value_executed` and `quantity_executed` as two independent + /// `Relaxed` loads, so under concurrent recording the ratio can be + /// transiently inconsistent (a torn read across the two counters); it + /// self-corrects once recording quiesces. #[must_use] pub fn average_execution_price(&self) -> Option { let qty = self.quantity_executed.load(Ordering::Relaxed); @@ -281,7 +366,12 @@ impl PriceLevelStatistics { } } - /// Get average waiting time for executed orders (in milliseconds) + /// Get average waiting time for executed orders (in milliseconds). + /// + /// Reads `sum_waiting_time` and `orders_executed` as two independent + /// `Relaxed` loads, so under concurrent recording the ratio can be + /// transiently inconsistent (a torn read across the two counters); it + /// self-corrects once recording quiesces. #[must_use] pub fn average_waiting_time(&self) -> Option { let count = self.orders_executed.load(Ordering::Relaxed); @@ -306,7 +396,17 @@ impl PriceLevelStatistics { } } - /// Reset all statistics + /// Reset all statistics to zero (and re-stamp `first_arrival_time`). + /// + /// # Quiescence contract + /// + /// This must only be called on a **quiescent** level — with no in-flight + /// [`record_execution`](Self::record_execution) (and hence no in-flight + /// `PriceLevel::match_order`). `reset` `store(0)`s each counter directly; a + /// concurrent `record_execution` overflow rollback (`fetch_sub`) racing that + /// `store(0)` would wrap a counter toward `u64::MAX`. No engine path calls + /// `reset` during matching, so this does not occur today, but it is a + /// caller obligation because `reset` is public. pub fn reset(&self) { let current_time = Self::current_timestamp_milliseconds_or_zero(); @@ -319,6 +419,7 @@ impl PriceLevelStatistics { self.first_arrival_time .store(current_time, Ordering::Relaxed); self.sum_waiting_time.store(0, Ordering::Relaxed); + self.stats_degraded.store(false, Ordering::Relaxed); } } @@ -348,6 +449,7 @@ impl Clone for PriceLevelStatistics { last_execution_time: AtomicU64::new(self.last_execution_time.load(Ordering::Relaxed)), first_arrival_time: AtomicU64::new(self.first_arrival_time.load(Ordering::Relaxed)), sum_waiting_time: AtomicU64::new(self.sum_waiting_time.load(Ordering::Relaxed)), + stats_degraded: AtomicBool::new(self.stats_degraded.load(Ordering::Relaxed)), } } } @@ -356,7 +458,7 @@ impl fmt::Display for PriceLevelStatistics { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!( f, - "PriceLevelStatistics:orders_added={};orders_removed={};orders_executed={};quantity_executed={};value_executed={};last_execution_time={};first_arrival_time={};sum_waiting_time={}", + "PriceLevelStatistics:orders_added={};orders_removed={};orders_executed={};quantity_executed={};value_executed={};last_execution_time={};first_arrival_time={};sum_waiting_time={};stats_degraded={}", self.orders_added.load(Ordering::Relaxed), self.orders_removed.load(Ordering::Relaxed), self.orders_executed.load(Ordering::Relaxed), @@ -364,7 +466,8 @@ impl fmt::Display for PriceLevelStatistics { self.value_executed.load(Ordering::Relaxed), self.last_execution_time.load(Ordering::Relaxed), self.first_arrival_time.load(Ordering::Relaxed), - self.sum_waiting_time.load(Ordering::Relaxed) + self.sum_waiting_time.load(Ordering::Relaxed), + self.stats_degraded.load(Ordering::Relaxed) ) } } @@ -438,6 +541,20 @@ impl FromStr for PriceLevelStatistics { let sum_waiting_time_str = get_field("sum_waiting_time")?; let sum_waiting_time = parse_u64("sum_waiting_time", sum_waiting_time_str)?; + // `stats_degraded` is optional for backward compatibility: a string + // produced before the field existed decodes with the flag cleared. + let stats_degraded = match fields.get("stats_degraded") { + Some(value) => { + value + .parse::() + .map_err(|_| PriceLevelError::InvalidFieldValue { + field: "stats_degraded".to_string(), + value: (*value).to_string(), + })? + } + None => false, + }; + Ok(PriceLevelStatistics { orders_added: AtomicUsize::new(orders_added), orders_removed: AtomicUsize::new(orders_removed), @@ -447,6 +564,7 @@ impl FromStr for PriceLevelStatistics { last_execution_time: AtomicU64::new(last_execution_time), first_arrival_time: AtomicU64::new(first_arrival_time), sum_waiting_time: AtomicU64::new(sum_waiting_time), + stats_degraded: AtomicBool::new(stats_degraded), }) } } @@ -456,7 +574,17 @@ impl Serialize for PriceLevelStatistics { where S: Serializer, { - let mut state = serializer.serialize_struct("PriceLevelStatistics", 8)?; + // Serialize `stats_degraded` ONLY when it is `true` (dynamic 8/9-field + // count). A non-degraded level then serializes in the exact pre-#117 + // 8-field form — byte-identical to a v2 statistics payload persisted + // before this flag existed — so a `PriceLevelSnapshotPackage`'s SHA-256 + // checksum, recomputed over the re-serialized bytes on + // `validate` / `from_snapshot_json`, still matches. A degraded level + // adds the field; `Deserialize` / `FromStr` default a missing flag to + // `false`, so both directions round-trip. + let degraded = self.stats_degraded.load(Ordering::Relaxed); + let field_count = if degraded { 9 } else { 8 }; + let mut state = serializer.serialize_struct("PriceLevelStatistics", field_count)?; state.serialize_field("orders_added", &self.orders_added.load(Ordering::Relaxed))?; state.serialize_field( @@ -487,6 +615,9 @@ impl Serialize for PriceLevelStatistics { "sum_waiting_time", &self.sum_waiting_time.load(Ordering::Relaxed), )?; + if degraded { + state.serialize_field("stats_degraded", &true)?; + } state.end() } @@ -506,6 +637,7 @@ impl<'de> Deserialize<'de> for PriceLevelStatistics { LastExecutionTime, FirstArrivalTime, SumWaitingTime, + StatsDegraded, } impl<'de> Deserialize<'de> for Field { @@ -535,6 +667,7 @@ impl<'de> Deserialize<'de> for PriceLevelStatistics { "last_execution_time" => Ok(Field::LastExecutionTime), "first_arrival_time" => Ok(Field::FirstArrivalTime), "sum_waiting_time" => Ok(Field::SumWaitingTime), + "stats_degraded" => Ok(Field::StatsDegraded), _ => Err(de::Error::unknown_field(value, FIELDS)), } } @@ -565,6 +698,7 @@ impl<'de> Deserialize<'de> for PriceLevelStatistics { let mut last_execution_time = None; let mut first_arrival_time = None; let mut sum_waiting_time = None; + let mut stats_degraded = None; while let Some(key) = map.next_key()? { match key { @@ -616,6 +750,12 @@ impl<'de> Deserialize<'de> for PriceLevelStatistics { } sum_waiting_time = Some(map.next_value()?); } + Field::StatsDegraded => { + if stats_degraded.is_some() { + return Err(de::Error::duplicate_field("stats_degraded")); + } + stats_degraded = Some(map.next_value()?); + } } } @@ -631,6 +771,9 @@ impl<'de> Deserialize<'de> for PriceLevelStatistics { }); let sum_waiting_time = sum_waiting_time.unwrap_or(0); + // Optional for backward compatibility: a payload written before + // the field existed decodes with the flag cleared. + let stats_degraded = stats_degraded.unwrap_or(false); Ok(PriceLevelStatistics { orders_added: AtomicUsize::new(orders_added), @@ -641,6 +784,7 @@ impl<'de> Deserialize<'de> for PriceLevelStatistics { last_execution_time: AtomicU64::new(last_execution_time), first_arrival_time: AtomicU64::new(first_arrival_time), sum_waiting_time: AtomicU64::new(sum_waiting_time), + stats_degraded: AtomicBool::new(stats_degraded), }) } } @@ -654,6 +798,7 @@ impl<'de> Deserialize<'de> for PriceLevelStatistics { "last_execution_time", "first_arrival_time", "sum_waiting_time", + "stats_degraded", ]; deserializer.deserialize_struct("PriceLevelStatistics", FIELDS, StatisticsVisitor) diff --git a/src/price_level/tests/level.rs b/src/price_level/tests/level.rs index 16eb00a..3ab2f41 100644 --- a/src/price_level/tests/level.rs +++ b/src/price_level/tests/level.rs @@ -7940,6 +7940,109 @@ mod tests { assert_counters_match_queue(&level); } } + + // ------------------------------------------------------------------ + // Issue #117 — statistics overflow degrades but never fails the trade + // ------------------------------------------------------------------ + + #[test] + fn test_match_order_stats_overflow_degrades_but_trade_intact() { + // Restore a level whose stats have quantity_executed near u64::MAX (no + // direct counter setter — seed through a snapshot). + let stats_text = format!( + "PriceLevelStatistics:orders_added=0;orders_removed=0;orders_executed=0;\ + quantity_executed={};value_executed=0;last_execution_time=0;first_arrival_time=0;\ + sum_waiting_time=0;stats_degraded=false", + u64::MAX - 5 + ); + let stats = crate::price_level::PriceLevelStatistics::from_str(&stats_text) + .expect("seed stats must parse"); + let order = std::sync::Arc::new(create_standard_order(1, 10_000, 100)); + let snapshot = crate::price_level::PriceLevelSnapshot::with_orders_and_stats( + Price::new(10_000), + vec![order], + stats, + ) + .expect("snapshot construction"); + let level = PriceLevel::from_snapshot(snapshot).expect("restore level"); + assert!( + !level.stats().stats_degraded(), + "precondition: not degraded yet" + ); + + // Match consumes maker 1 (quantity 100); recording quantity += 100 over + // u64::MAX - 5 overflows, so the execution's stats are dropped. + let generator = + UuidGenerator::new(Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap()); + let result = level.match_order( + 100, + Id::from_u64(999), + TimeInForce::Gtc, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ); + + // The trade is still emitted and the MatchResult is intact. + assert_eq!(result.trades().len(), 1, "trade must still be emitted"); + assert_eq!( + result.executed_quantity().expect("no overflow").as_u64(), + 100 + ); + assert_eq!( + result.trades().as_vec()[0].maker_order_id(), + Id::from_u64(1) + ); + + // Stats degraded, and the aggregate did NOT advance (all-or-nothing). + assert!( + level.stats().stats_degraded(), + "stats must be marked degraded" + ); + assert_eq!( + level.stats().quantity_executed(), + u64::MAX - 5, + "quantity_executed unchanged (execution dropped)" + ); + assert_eq!( + level.stats().orders_executed(), + 0, + "orders_executed rolled back" + ); + + // The degraded flag round-trips through a snapshot. + let json = level.snapshot_to_json().expect("snapshot json"); + let restored = PriceLevel::from_snapshot_json(&json).expect("restore json"); + assert!( + restored.stats().stats_degraded(), + "the degraded flag must round-trip through a snapshot" + ); + } + + #[test] + fn test_snapshot_v2_without_degraded_flag_validates_and_roundtrips() { + // A non-degraded level serializes its statistics in the pre-#117 + // 8-field form (the flag is skipped when false), so its checksummed + // package is byte-compatible with a v2 package persisted before the + // flag existed: the SHA-256 recomputed over those 8-field bytes on + // `from_snapshot_json` still validates. This guards against the + // "old snapshot fails ChecksumMismatch after upgrade" regression. + let level = PriceLevel::new(10_000); + level + .add_order(create_standard_order(1, 10_000, 100)) + .expect("seed maker"); + let json = level.snapshot_to_json().expect("snapshot json"); + assert!( + !json.contains("stats_degraded"), + "a non-degraded snapshot must omit the flag (old-v2 checksum compat)" + ); + // The package validates (checksum over the 8-field statistics bytes) and + // restores — i.e. an old-format fixture passes `validate`. + let restored = + PriceLevel::from_snapshot_json(&json).expect("old-format package must validate"); + assert!(!restored.stats().stats_degraded()); + assert_eq!(restored.order_count(), 1); + } } #[cfg(test)] diff --git a/src/price_level/tests/statistics.rs b/src/price_level/tests/statistics.rs index d824d75..07f076b 100644 --- a/src/price_level/tests/statistics.rs +++ b/src/price_level/tests/statistics.rs @@ -420,5 +420,182 @@ mod tests { assert_eq!(deserialized.orders_executed(), 3); assert_eq!(deserialized.quantity_executed(), 0); assert_eq!(deserialized.value_executed(), 0); + // Missing stats_degraded defaults to false. + assert!(!deserialized.stats_degraded()); + } + + // ------------------------------------------------------------------ + // Issue #117 — all-or-nothing statistics + degraded flag + // ------------------------------------------------------------------ + + /// Build statistics pre-loaded with specific aggregate values (no public + /// counter setter exists, so seed via the `FromStr` surface). + fn seed_stats( + orders_executed: usize, + quantity_executed: u64, + value_executed: u64, + sum_waiting_time: u64, + ) -> PriceLevelStatistics { + let text = format!( + "PriceLevelStatistics:orders_added=0;orders_removed=0;orders_executed={orders_executed};\ + quantity_executed={quantity_executed};value_executed={value_executed};\ + last_execution_time=0;first_arrival_time=0;sum_waiting_time={sum_waiting_time};\ + stats_degraded=false" + ); + PriceLevelStatistics::from_str(&text).expect("seed stats must parse") + } + + #[test] + fn test_record_execution_all_or_nothing_on_each_overflow() { + // quantity_executed overflow: nothing advances, degraded set. + { + let stats = seed_stats(0, u64::MAX - 5, 0, 0); + assert!(stats.record_execution(10, 1, 0, 1_000).is_err()); + assert_eq!(stats.orders_executed(), 0, "orders_executed rolled back"); + assert_eq!( + stats.quantity_executed(), + u64::MAX - 5, + "quantity unchanged" + ); + assert_eq!(stats.value_executed(), 0, "value unchanged"); + assert!(stats.stats_degraded(), "degraded flag set"); + } + // value_executed overflow: orders + quantity rolled back. + { + let stats = seed_stats(0, 0, u64::MAX - 5, 0); + assert!(stats.record_execution(1, 10, 0, 1_000).is_err()); + assert_eq!(stats.orders_executed(), 0); + assert_eq!(stats.quantity_executed(), 0, "quantity rolled back"); + assert_eq!(stats.value_executed(), u64::MAX - 5, "value unchanged"); + assert!(stats.stats_degraded()); + } + // sum_waiting_time overflow: orders + quantity + value rolled back. + { + let stats = seed_stats(0, 0, 0, u64::MAX - 5); + assert!(stats.record_execution(1, 1, 1, 100).is_err()); + assert_eq!(stats.orders_executed(), 0); + assert_eq!(stats.quantity_executed(), 0, "quantity rolled back"); + assert_eq!(stats.value_executed(), 0, "value rolled back"); + assert_eq!(stats.sum_waiting_time(), u64::MAX - 5, "sum unchanged"); + assert!(stats.stats_degraded()); + } + // value multiplication exceeds u64 storage (validation, before mutation). + { + let stats = seed_stats(0, 0, 0, 0); + assert!(stats.record_execution(u64::MAX, 2, 0, 1_000).is_err()); + assert_eq!(stats.orders_executed(), 0); + assert_eq!(stats.quantity_executed(), 0); + assert_eq!(stats.value_executed(), 0); + assert!(stats.stats_degraded()); + } + // Maker timestamp in the future of execution (validation). + { + let stats = seed_stats(0, 0, 0, 0); + assert!(stats.record_execution(1, 1, 200, 100).is_err()); + assert_eq!(stats.orders_executed(), 0); + assert!(stats.stats_degraded()); + } + } + + #[test] + fn test_record_execution_success_leaves_flag_clear() { + let stats = PriceLevelStatistics::new(); + assert!(stats.record_execution(10, 5, 0, 1_000).is_ok()); + assert_eq!(stats.orders_executed(), 1); + assert_eq!(stats.quantity_executed(), 10); + assert_eq!(stats.value_executed(), 50); + assert!( + !stats.stats_degraded(), + "a successful record must not degrade" + ); + } + + #[test] + fn test_concurrent_record_execution_cross_aggregate_consistency() { + use std::sync::Barrier; + + // Seed quantity_executed AND value_executed with exactly K*Q of headroom + // below u64::MAX (SYMMETRIC headroom, price = 1, so each record adds Q to + // both). With N > K concurrent records, exactly K fit and the rest fail + // at the FIRST additive counter (quantity_executed) — so in this + // symmetric config a loser never advances any counter and no + // cross-counter rollback is exercised; it is a clean reject. (Under + // ASYMMETRIC headroom a loser could pass quantity and fail at value, and + // the exact-fit count would be `<= K` rather than `== K`; the rollback + // path is covered by the deterministic per-aggregate test above.) + // Invariant here: orders_executed == K, quantity_executed == + // value_executed (each success advances both in lockstep), the surplus + // over the seed equals orders_executed * Q, and the degraded flag is set. + const N: usize = 8; + const K: u64 = 3; + const Q: u64 = 1_000; + let seed = u64::MAX - K * Q; + + for _ in 0..50 { + let stats = Arc::new(seed_stats(0, seed, seed, 0)); + // Barrier so all N records race from the same instant. + let barrier = Arc::new(Barrier::new(N)); + let mut handles = Vec::with_capacity(N); + for _ in 0..N { + let stats = Arc::clone(&stats); + let barrier = Arc::clone(&barrier); + handles.push(thread::spawn(move || { + barrier.wait(); + stats.record_execution(Q, 1, 0, 1_000).is_ok() + })); + } + let successes: u64 = handles + .into_iter() + .map(|h| u64::from(h.join().expect("thread panicked"))) + .sum(); + + assert_eq!(successes, K, "exactly K records must fit the headroom"); + assert_eq!(stats.orders_executed() as u64, K); + assert_eq!( + stats.quantity_executed(), + stats.value_executed(), + "quantity and value advance in lockstep (each success adds to both)" + ); + assert_eq!( + stats.quantity_executed() - seed, + K * Q, + "the surplus over the seed equals orders_executed * Q" + ); + assert!( + stats.stats_degraded(), + "some records overflowed -> degraded" + ); + } + } + + #[test] + fn test_serialize_omits_degraded_flag_when_false() { + // A non-degraded statistics serializes in the pre-#117 8-field form + // (the flag is skipped when false), so a v2 snapshot payload persisted + // before the flag existed re-serializes byte-identically and its SHA-256 + // checksum still validates. + let stats = PriceLevelStatistics::new(); + assert!(!stats.stats_degraded()); + let json = serde_json::to_string(&stats).expect("serialize"); + assert!( + !json.contains("stats_degraded"), + "a non-degraded statistics must omit the flag (old-v2 checksum compat): {json}" + ); + // It still round-trips: a missing flag decodes to false. + let back: PriceLevelStatistics = serde_json::from_str(&json).expect("deserialize"); + assert!(!back.stats_degraded()); + + // A degraded statistics DOES emit the field, and it round-trips. + let degraded = seed_stats(0, 0, 0, 0); + assert!(degraded.record_execution(u64::MAX, 2, 0, 1_000).is_err()); // value overflow + assert!(degraded.stats_degraded()); + let degraded_json = serde_json::to_string(°raded).expect("serialize degraded"); + assert!( + degraded_json.contains("\"stats_degraded\":true"), + "a degraded statistics must emit the flag: {degraded_json}" + ); + let degraded_back: PriceLevelStatistics = + serde_json::from_str(°raded_json).expect("deserialize degraded"); + assert!(degraded_back.stats_degraded(), "the flag must round-trip"); } } From 05cda8e699b57e1200d98dc12dcde1a330187710 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 18:35:16 +0200 Subject: [PATCH 2/4] address review: statistics seqlock, format v3, monotonic timestamp - an AtomicU64 sequence brackets every statistics WRITE (record_execution and reset, via a RAII guard that closes the section on every early return) and Clone / Serialize / Display retry their multi-field copy until no write overlaps: a checksummed snapshot can no longer capture an in-flight recording prefix, and reset can no longer interleave inside a rollback (it is a seqlock writer now, enforced rather than documented); reader liveness is argued from the writer-serialization contract and the guard's panic-safe Drop - orders_executed joins the checked all-or-nothing transaction (a FromStr-seeded usize::MAX rejects instead of wrapping while the other aggregates advance) - the degraded WARN fires only on the false-to-true transition of the sticky flag (compare_exchange decides the witness) -- a burst of drops logs once - snapshot format bumps to v3: new packages write v3, validate() accepts v2 (legacy 8-field statistics, flag defaults false) and rejects v1; statistics keep skip-when-false serialization so checksum recomputation stays version-agnostic and legacy v2 checksums still match; migration guide + README updated - last_execution_time commits via fetch_max, so an out-of-order recorder can never move the latest-execution timestamp backwards --- README.md | 17 + .../src/bin/integration_snapshot_recovery.rs | 5 +- src/lib.rs | 17 + src/price_level/level.rs | 12 +- src/price_level/snapshot.rs | 33 +- src/price_level/statistics.rs | 352 +++++++++++++----- src/price_level/tests/snapshot.rs | 81 ++++ src/price_level/tests/statistics.rs | 154 ++++++++ 8 files changed, 567 insertions(+), 104 deletions(-) diff --git a/README.md b/README.md index 2ff0488..79a1b35 100644 --- a/README.md +++ b/README.md @@ -452,6 +452,23 @@ checksum error). Re-take any persisted snapshots with this release. No code changes are required at the call sites: `snapshot_to_json()` / `from_snapshot_json()` keep the same signatures. +### Migration Guide (snapshot format v2 → v3) + +`SNAPSHOT_FORMAT_VERSION` is bumped from `2` to `3` (issue #129). Version 3 +owns the optional 9th statistics field, `stats_degraded` (issue #117): a +**degraded** level — one where an execution's statistics contribution was +dropped all-or-nothing — serializes that field, and such a payload is now a +v3 package rather than a v2 package mislabelled with an extra field an old +8-field-only reader would reject. + +Restore is **backward compatible**: [`PriceLevelSnapshotPackage::validate`] +accepts **both** v2 (legacy, 8-field statistics, `stats_degraded` defaults +`false`) and v3, so snapshots written by the previous release keep restoring +unchanged; only v1 is still rejected. Checksum recomputation is +version-agnostic — a non-degraded level serializes the same 8 fields under +either version, so a legacy v2 package's SHA-256 still matches. New snapshots +are written at v3. No code changes are required at the call sites. + ### Migration Guide (`Trade::total_value` is now checked) [`Trade::total_value`](crate::execution::Trade::total_value) now returns diff --git a/examples/src/bin/integration_snapshot_recovery.rs b/examples/src/bin/integration_snapshot_recovery.rs index 28ffd59..b6b1c22 100644 --- a/examples/src/bin/integration_snapshot_recovery.rs +++ b/examples/src/bin/integration_snapshot_recovery.rs @@ -86,8 +86,9 @@ fn main() { .snapshot_package() .unwrap_or_else(|e| exit_err(&format!("snapshot_package: {e}"))); - // Snapshot format v2 (statistics persisted, issue #63). - assert_eq_or_exit(package.version(), 2, "snapshot version"); + // Snapshot format v3 (statistics persisted since v2/#63; v3 owns the + // optional `stats_degraded` field, issue #129). + assert_eq_or_exit(package.version(), 3, "snapshot version"); assert_or_exit( !package.checksum().is_empty(), "checksum should not be empty", diff --git a/src/lib.rs b/src/lib.rs index c6e9dfd..f2e16f5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -444,6 +444,23 @@ //! changes are required at the call sites: `snapshot_to_json()` / //! `from_snapshot_json()` keep the same signatures. //! +//! ## Migration Guide (snapshot format v2 → v3) +//! +//! `SNAPSHOT_FORMAT_VERSION` is bumped from `2` to `3` (issue #129). Version 3 +//! owns the optional 9th statistics field, `stats_degraded` (issue #117): a +//! **degraded** level — one where an execution's statistics contribution was +//! dropped all-or-nothing — serializes that field, and such a payload is now a +//! v3 package rather than a v2 package mislabelled with an extra field an old +//! 8-field-only reader would reject. +//! +//! Restore is **backward compatible**: [`PriceLevelSnapshotPackage::validate`] +//! accepts **both** v2 (legacy, 8-field statistics, `stats_degraded` defaults +//! `false`) and v3, so snapshots written by the previous release keep restoring +//! unchanged; only v1 is still rejected. Checksum recomputation is +//! version-agnostic — a non-degraded level serializes the same 8 fields under +//! either version, so a legacy v2 package's SHA-256 still matches. New snapshots +//! are written at v3. No code changes are required at the call sites. +//! //! ## Migration Guide (`Trade::total_value` is now checked) //! //! [`Trade::total_value`](crate::execution::Trade::total_value) now returns diff --git a/src/price_level/level.rs b/src/price_level/level.rs index 00e550d..e7327b0 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -1386,12 +1386,22 @@ impl PriceLevel { // `PriceLevelStatistics::stats_degraded`), leaving the // trade stream unaffected. Log the drop rather than // discard it silently. + // Read the degraded flag BEFORE recording: under the + // single-matcher-per-level model this is the faithful + // witness of the `false -> true` transition (issue #129). + // `record_execution` sets the flag atomically via + // `compare_exchange`; we log the WARN only on the FIRST + // drop that transitions it, so a burst of dropped + // executions logs once, not once per drop — the sticky + // flag remains the durable signal for the rest. + let was_degraded = self.stats.stats_degraded(); if let Err(err) = self.stats.record_execution( data.consumed, data.maker_price, data.maker_timestamp, timestamp.as_u64(), - ) { + ) && !was_degraded + { // WARN, not ERROR: the match is not aborted — this is // a recoverable observability anomaly flagged by the // sticky degraded flag (the trade is committed). diff --git a/src/price_level/snapshot.rs b/src/price_level/snapshot.rs index bac178f..ab2a162 100644 --- a/src/price_level/snapshot.rs +++ b/src/price_level/snapshot.rs @@ -261,12 +261,29 @@ impl PriceLevelSnapshot { } } -/// Format version for checksum-enabled price level snapshots. +/// Format version for checksum-enabled price level snapshots. New packages are +/// written at this version. /// -/// Version 2 (issue #63) persists per-level [`PriceLevelStatistics`] inside the -/// snapshot. Version 1 packages carried no statistics and are rejected by -/// [`PriceLevelSnapshotPackage::validate`] with a version mismatch. -pub const SNAPSHOT_FORMAT_VERSION: u32 = 2; +/// - **Version 1** carried no statistics — rejected by +/// [`PriceLevelSnapshotPackage::validate`] with a version mismatch. +/// - **Version 2** (issue #63) persists per-level [`PriceLevelStatistics`] as an +/// 8-field statistics payload (no `stats_degraded`). +/// - **Version 3** (issue #129) is the current shape: it owns the optional 9th +/// `stats_degraded` statistics field. A degraded level (which serializes that +/// field) is a v3 payload, so it is no longer mislabelled v2 where an old +/// 8-field-only reader would choke on the unknown field. +/// +/// [`PriceLevelSnapshotPackage::validate`] accepts BOTH v2 (legacy, 8-field, +/// `stats_degraded` defaults `false`) and v3, so old snapshots keep restoring; +/// v1 is still rejected. Checksum recomputation is version-agnostic — a +/// non-degraded level serializes 8 fields under either version, so a legacy v2 +/// package's SHA-256 still matches. +pub const SNAPSHOT_FORMAT_VERSION: u32 = 3; + +/// The set of snapshot format versions [`PriceLevelSnapshotPackage::validate`] +/// accepts on restore: the current [`SNAPSHOT_FORMAT_VERSION`] (v3) and the +/// legacy v2 (issue #129). v1 (statistics-less) is not accepted. +const SUPPORTED_SNAPSHOT_VERSIONS: &[u32] = &[2, 3]; /// Serialized representation of a price level snapshot including checksum validation metadata. /// @@ -361,11 +378,11 @@ impl PriceLevelSnapshotPackage { // Snapshot restoration / validation is a cold path: keep it out of line. #[inline(never)] pub fn validate(&self) -> Result<(), PriceLevelError> { - if self.version != SNAPSHOT_FORMAT_VERSION { + if !SUPPORTED_SNAPSHOT_VERSIONS.contains(&self.version) { return Err(PriceLevelError::InvalidOperation { message: format!( - "Unsupported snapshot version: {} (expected {})", - self.version, SNAPSHOT_FORMAT_VERSION + "Unsupported snapshot version: {} (expected one of {:?})", + self.version, SUPPORTED_SNAPSHOT_VERSIONS ), }); } diff --git a/src/price_level/statistics.rs b/src/price_level/statistics.rs index 65d75fc..c101235 100644 --- a/src/price_level/statistics.rs +++ b/src/price_level/statistics.rs @@ -18,22 +18,26 @@ use std::time::{SystemTime, UNIX_EPOCH}; /// /// # Atomic ordering /// -/// **Every atomic access in this type uses [`Ordering::Relaxed`], deliberately.** -/// These are pure observability counters: nothing in the engine reads a -/// statistic to gate a queue mutation, and no other field's visibility is -/// published through them. They carry no happens-before relationship — each -/// counter is independent and may be read mid-update, so a multi-field read -/// (serde, [`Display`](std::fmt::Display), [`Clone`]) is an explicitly -/// best-effort, non-transactional point-in-time view that can be torn across -/// fields. This is unlike `PriceLevel::snapshot`, which is internally -/// consistent (it derives its aggregates from one materialized order vector); -/// these statistics counters are not part of that consistency guarantee. -/// `Relaxed` is therefore both correct and the -/// cheapest valid choice; a stronger ordering would only add cost without -/// buying any guarantee a consumer relies on. The lone read-modify-write loop -/// in [`checked_fetch_add_u64`](Self::checked_fetch_add_u64) is a standard -/// `compare_exchange_weak` CAS retry — see the note there for why both its -/// success and failure orderings are `Relaxed`. +/// The individual counters are `Relaxed` observability atomics: nothing in the +/// engine reads a statistic to gate a queue mutation, and no other field's +/// visibility is published through them. Point accessors ([`orders_executed`](Self::orders_executed), +/// the average ratios, …) load them `Relaxed` and remain best-effort — a ratio +/// across two counters can still be transiently torn and self-corrects once +/// recording quiesces. +/// +/// A **multi-field** read — [`Clone`] (which backs the checksummed +/// `PriceLevel::snapshot`) and the serde / [`Display`](std::fmt::Display) +/// serialization paths — instead goes through a **seqlock** (issue #129) so it +/// copies a consistent set that never mixes a pre- and post-`record_execution` +/// prefix (which would otherwise checksum a state the level never held). The +/// `stats_seq` sequence counter is bumped to odd on a writer's entry +/// ([`record_execution`](Self::record_execution) / [`reset`](Self::reset)) and +/// back to even on its exit, both `Release`; a reader loads it `Acquire`, copies +/// the fields, `Acquire`-fences, re-loads it, and retries if it changed or was +/// odd. Writers are serialized by the engine model (one matcher per level + +/// `reset`'s quiescence contract), which the seqlock assumes. The lone +/// read-modify-write loop in [`checked_fetch_add_u64`](Self::checked_fetch_add_u64) +/// is a standard `compare_exchange_weak` CAS retry. #[derive(Debug)] pub struct PriceLevelStatistics { /// Number of orders added @@ -69,6 +73,58 @@ pub struct PriceLevelStatistics { /// is the observable signal that the recorded aggregates under-count the /// true executions. Serialized so it round-trips through a snapshot. stats_degraded: AtomicBool, + + /// Seqlock sequence for consistent multi-field reads (issue #129). Even when + /// no writer is in a section, odd while a writer ([`record_execution`](Self::record_execution) + /// / [`reset`](Self::reset)) is mutating. Purely internal — never serialized + /// — so a restored / cloned value starts even (0). + stats_seq: AtomicU64, +} + +/// RAII guard bracketing a statistics WRITE section for the seqlock (issue +/// #129). Constructing it bumps `stats_seq` to odd; dropping it bumps back to +/// even, so a concurrent multi-field reader retries if it overlapped either +/// increment. Using a guard keeps the section correct across the early returns +/// in [`PriceLevelStatistics::record_execution`]. +struct WriteSeqGuard<'a> { + seq: &'a AtomicU64, +} + +impl<'a> WriteSeqGuard<'a> { + #[inline] + fn new(seq: &'a AtomicU64) -> Self { + // Enter: even -> odd. `Relaxed` RMW plus a `Release` fence so the field + // writes that follow cannot be reordered before the odd marker a reader + // watches for. + seq.fetch_add(1, Ordering::Relaxed); + std::sync::atomic::fence(Ordering::Release); + Self { seq } + } +} + +impl Drop for WriteSeqGuard<'_> { + #[inline] + fn drop(&mut self) { + // Exit: odd -> even, `Release` so every field write in the section + // happens-before a reader's `Acquire` load of the (now even) sequence. + self.seq.fetch_add(1, Ordering::Release); + } +} + +/// A consistent point-in-time copy of every statistics field, read under the +/// seqlock (issue #129). Plain values, no atomics — so `Clone` / serialize +/// materialize a coherent set rather than a torn mix of counters. +#[derive(Clone, Copy)] +struct StatsData { + orders_added: usize, + orders_removed: usize, + orders_executed: usize, + quantity_executed: u64, + value_executed: u64, + last_execution_time: u64, + first_arrival_time: u64, + sum_waiting_time: u64, + stats_degraded: bool, } impl PriceLevelStatistics { @@ -104,6 +160,108 @@ impl PriceLevelStatistics { } } + /// Checked `+= value` on a `usize` counter, mirroring + /// [`checked_fetch_add_u64`](Self::checked_fetch_add_u64). `orders_executed` + /// can be seeded to `usize::MAX` through `FromStr` / serde, so a plain + /// `fetch_add(1)` could wrap while the other aggregates advance (issue #129); + /// this rejects the overflow so the all-or-nothing rollback can undo the + /// prefix instead. + fn checked_fetch_add_usize( + target: &AtomicUsize, + value: usize, + field: &str, + ) -> Result<(), PriceLevelError> { + let mut current = target.load(Ordering::Relaxed); + loop { + let next = + current + .checked_add(value) + .ok_or_else(|| PriceLevelError::InvalidOperation { + message: format!("{field} overflow"), + })?; + match target.compare_exchange_weak(current, next, Ordering::Relaxed, Ordering::Relaxed) + { + Ok(_) => return Ok(()), + Err(observed) => current = observed, + } + } + } + + /// Set the sticky degraded flag; returns `true` iff THIS call transitioned it + /// `false -> true` (issue #129). The caller (`PriceLevel::match_order`) logs + /// the WARN only on that transition, so a burst of dropped executions marks + /// the level degraded once and logs once, not once per drop. + pub(crate) fn mark_degraded(&self) -> bool { + self.stats_degraded + .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + /// Read a consistent snapshot of every field under the seqlock (issue #129). + /// + /// Retries until a full copy brackets an even, unchanged sequence — i.e. no + /// writer transaction ([`record_execution`](Self::record_execution) / + /// [`reset`](Self::reset)) overlapped it, so the copy is NEVER a torn mix of + /// a pre- and post-write prefix (a bounded fallback that returned a torn copy + /// would defeat the checksummed snapshot this backs). + /// + /// # Liveness + /// + /// The loop's work PER attempt is bounded (one sequence load + a nine-field + /// copy), and it converges under the advisory writer-serialization contract + /// (one matcher per level + `reset`'s quiescence): a writer holds the section + /// for only a short, allocation-free burst before dropping the guard back to + /// even, and `record_execution` is finite, so a reader exits on the first + /// iteration when uncontended and otherwise as soon as recording quiesces + /// (which it always does — the matcher cannot record forever). A panicking + /// writer still restores the even sequence via the guard's `Drop`, so the + /// reader is never stranded on a permanently-odd sequence. + fn read_consistent(&self) -> StatsData { + loop { + let s1 = self.stats_seq.load(Ordering::Acquire); + if s1 & 1 != 0 { + // A writer is mid-section; spin until it exits. + std::hint::spin_loop(); + continue; + } + let data = StatsData { + orders_added: self.orders_added.load(Ordering::Relaxed), + orders_removed: self.orders_removed.load(Ordering::Relaxed), + orders_executed: self.orders_executed.load(Ordering::Relaxed), + quantity_executed: self.quantity_executed.load(Ordering::Relaxed), + value_executed: self.value_executed.load(Ordering::Relaxed), + last_execution_time: self.last_execution_time.load(Ordering::Relaxed), + first_arrival_time: self.first_arrival_time.load(Ordering::Relaxed), + sum_waiting_time: self.sum_waiting_time.load(Ordering::Relaxed), + stats_degraded: self.stats_degraded.load(Ordering::Relaxed), + }; + // Ensure the field loads complete before re-reading the sequence. + std::sync::atomic::fence(Ordering::Acquire); + let s2 = self.stats_seq.load(Ordering::Relaxed); + if s1 == s2 { + return data; + } + std::hint::spin_loop(); + } + } + + /// Reconstruct from a plain [`StatsData`] copy (seqlock reader output), with + /// a fresh even sequence. + fn from_data(data: StatsData) -> Self { + Self { + orders_added: AtomicUsize::new(data.orders_added), + orders_removed: AtomicUsize::new(data.orders_removed), + orders_executed: AtomicUsize::new(data.orders_executed), + quantity_executed: AtomicU64::new(data.quantity_executed), + value_executed: AtomicU64::new(data.value_executed), + last_execution_time: AtomicU64::new(data.last_execution_time), + first_arrival_time: AtomicU64::new(data.first_arrival_time), + sum_waiting_time: AtomicU64::new(data.sum_waiting_time), + stats_degraded: AtomicBool::new(data.stats_degraded), + stats_seq: AtomicU64::new(0), + } + } + #[inline] fn current_timestamp_milliseconds() -> Result { SystemTime::now() @@ -134,6 +292,7 @@ impl PriceLevelStatistics { first_arrival_time: AtomicU64::new(current_time), sum_waiting_time: AtomicU64::new(0), stats_degraded: AtomicBool::new(false), + stats_seq: AtomicU64::new(0), } } @@ -192,6 +351,13 @@ impl PriceLevelStatistics { ) -> Result<(), PriceLevelError> { let current_time = execution_timestamp; + // Bracket the whole record as a seqlock WRITE (issue #129): a concurrent + // multi-field reader (`Clone` / serialize) retries rather than capture an + // in-flight prefix, and `reset` — also a writer — cannot interleave this + // transaction's rollback. The guard's `Drop` closes the section (back to + // even) on EVERY return path below, including the early validation errors. + let _write = WriteSeqGuard::new(&self.stats_seq); + // Validate everything that can fail BEFORE mutating any counter, so a // rejected record leaves the statistics untouched. Any failure marks the // stats degraded: this execution's contribution is being dropped. @@ -199,7 +365,7 @@ impl PriceLevelStatistics { match current_time.checked_sub(order_timestamp) { Some(value) => Some(value), None => { - self.stats_degraded.store(true, Ordering::Relaxed); + self.mark_degraded(); return Err(PriceLevelError::InvalidOperation { message: format!( "order timestamp {order_timestamp} is in the future of current time {current_time}" @@ -217,7 +383,7 @@ impl PriceLevelStatistics { { Some(value) => value, None => { - self.stats_degraded.store(true, Ordering::Relaxed); + self.mark_degraded(); return Err(PriceLevelError::InvalidOperation { message: "value_executed overflow (quantity * price exceeds u64 storage)" .to_string(), @@ -227,17 +393,22 @@ impl PriceLevelStatistics { // Commit the additive aggregates with a rollback of the already-committed // prefix on a later overflow (all-or-nothing). `orders_executed` is a - // `usize` counter that cannot overflow in practice, but it is part of the - // transaction so it is rolled back too. `last_execution_time` is a - // non-additive "latest" store, applied only after every additive commit - // succeeds so a rejected record never advances it. - self.orders_executed.fetch_add(1, Ordering::Relaxed); + // `usize` counter that could be seeded to `usize::MAX` via FromStr / serde + // (issue #129), so it is a CHECKED add and part of the transaction — + // rolled back like the others. `last_execution_time` is a non-additive + // "latest" store, applied only after every additive commit succeeds so a + // rejected record never advances it. + if let Err(err) = Self::checked_fetch_add_usize(&self.orders_executed, 1, "orders_executed") + { + self.mark_degraded(); + return Err(err); + } if let Err(err) = Self::checked_fetch_add_u64(&self.quantity_executed, quantity, "quantity_executed") { self.orders_executed.fetch_sub(1, Ordering::Relaxed); - self.stats_degraded.store(true, Ordering::Relaxed); + self.mark_degraded(); return Err(err); } @@ -247,7 +418,7 @@ impl PriceLevelStatistics { self.quantity_executed .fetch_sub(quantity, Ordering::Relaxed); self.orders_executed.fetch_sub(1, Ordering::Relaxed); - self.stats_degraded.store(true, Ordering::Relaxed); + self.mark_degraded(); return Err(err); } @@ -262,12 +433,15 @@ impl PriceLevelStatistics { self.quantity_executed .fetch_sub(quantity, Ordering::Relaxed); self.orders_executed.fetch_sub(1, Ordering::Relaxed); - self.stats_degraded.store(true, Ordering::Relaxed); + self.mark_degraded(); return Err(err); } + // Monotonic (issue #129): a concurrent / out-of-order record can never + // move the "latest execution" backwards. Belt-and-braces with the + // seqlock, but cheap and independently correct. self.last_execution_time - .store(current_time, Ordering::Relaxed); + .fetch_max(current_time, Ordering::Relaxed); Ok(()) } @@ -402,14 +576,24 @@ impl PriceLevelStatistics { /// /// This must only be called on a **quiescent** level — with no in-flight /// [`record_execution`](Self::record_execution) (and hence no in-flight - /// `PriceLevel::match_order`). `reset` `store(0)`s each counter directly; a - /// concurrent `record_execution` overflow rollback (`fetch_sub`) racing that - /// `store(0)` would wrap a counter toward `u64::MAX`. No engine path calls - /// `reset` during matching, so this does not occur today, but it is a - /// caller obligation because `reset` is public. + /// `PriceLevel::match_order`). `reset` participates in the seqlock as a + /// WRITER (issue #129), so it can never interleave the middle of a + /// `record_execution` transaction's rollback (which would otherwise wrap a + /// counter toward `u64::MAX` via a `store(0)` racing a `fetch_sub`) — but the + /// seqlock protects READERS, it does not serialize two writers. The + /// single-matcher-per-level model already serializes `record_execution`, and + /// this quiescence requirement extends that to `reset`. No engine path calls + /// `reset` during matching, so the race does not occur today; it remains a + /// caller obligation because `reset` is public. A multi-field reader + /// (`Clone` / serialize) racing `reset` retries and observes either the + /// pre-reset or fully-reset state, never a mix. pub fn reset(&self) { let current_time = Self::current_timestamp_milliseconds_or_zero(); + // Seqlock write section: a concurrent multi-field reader retries rather + // than capture a half-reset copy. + let _write = WriteSeqGuard::new(&self.stats_seq); + self.orders_added.store(0, Ordering::Relaxed); self.orders_removed.store(0, Ordering::Relaxed); self.orders_executed.store(0, Ordering::Relaxed); @@ -430,44 +614,37 @@ impl Default for PriceLevelStatistics { } impl Clone for PriceLevelStatistics { - /// Clones the statistics by snapshotting each atomic counter. + /// Clones the statistics by reading a **consistent** snapshot of every field + /// under the seqlock (issue #129). /// - /// Each counter is read with [`Ordering::Relaxed`]: the clone is a - /// best-effort point-in-time copy of independent counters (the same - /// contract as the serde path and `snapshot()`), not a globally consistent - /// transactional read across all eight fields. This is the representation - /// persisted in [`PriceLevelSnapshot`](crate::price_level::PriceLevelSnapshot), - /// so a restored level carries the recorded statistics rather than a fresh, - /// zeroed set. + /// This is the representation persisted in + /// [`PriceLevelSnapshot`](crate::price_level::PriceLevelSnapshot) and hence + /// covered by that snapshot's SHA-256 checksum, so the copy must not mix a + /// pre- and post-`record_execution` prefix — the seqlock retries until it + /// captures a state the level actually held. A restored level therefore + /// carries the recorded statistics rather than a fresh, zeroed set. fn clone(&self) -> Self { - Self { - orders_added: AtomicUsize::new(self.orders_added.load(Ordering::Relaxed)), - orders_removed: AtomicUsize::new(self.orders_removed.load(Ordering::Relaxed)), - orders_executed: AtomicUsize::new(self.orders_executed.load(Ordering::Relaxed)), - quantity_executed: AtomicU64::new(self.quantity_executed.load(Ordering::Relaxed)), - value_executed: AtomicU64::new(self.value_executed.load(Ordering::Relaxed)), - last_execution_time: AtomicU64::new(self.last_execution_time.load(Ordering::Relaxed)), - first_arrival_time: AtomicU64::new(self.first_arrival_time.load(Ordering::Relaxed)), - sum_waiting_time: AtomicU64::new(self.sum_waiting_time.load(Ordering::Relaxed)), - stats_degraded: AtomicBool::new(self.stats_degraded.load(Ordering::Relaxed)), - } + Self::from_data(self.read_consistent()) } } impl fmt::Display for PriceLevelStatistics { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + // Consistent multi-field read (issue #129): the emitted string is a + // coherent snapshot, not a torn mix, and round-trips through `FromStr`. + let d = self.read_consistent(); write!( f, "PriceLevelStatistics:orders_added={};orders_removed={};orders_executed={};quantity_executed={};value_executed={};last_execution_time={};first_arrival_time={};sum_waiting_time={};stats_degraded={}", - self.orders_added.load(Ordering::Relaxed), - self.orders_removed.load(Ordering::Relaxed), - self.orders_executed.load(Ordering::Relaxed), - self.quantity_executed.load(Ordering::Relaxed), - self.value_executed.load(Ordering::Relaxed), - self.last_execution_time.load(Ordering::Relaxed), - self.first_arrival_time.load(Ordering::Relaxed), - self.sum_waiting_time.load(Ordering::Relaxed), - self.stats_degraded.load(Ordering::Relaxed) + d.orders_added, + d.orders_removed, + d.orders_executed, + d.quantity_executed, + d.value_executed, + d.last_execution_time, + d.first_arrival_time, + d.sum_waiting_time, + d.stats_degraded ) } } @@ -565,6 +742,7 @@ impl FromStr for PriceLevelStatistics { first_arrival_time: AtomicU64::new(first_arrival_time), sum_waiting_time: AtomicU64::new(sum_waiting_time), stats_degraded: AtomicBool::new(stats_degraded), + stats_seq: AtomicU64::new(0), }) } } @@ -574,47 +752,34 @@ impl Serialize for PriceLevelStatistics { where S: Serializer, { + // Read all fields as ONE consistent seqlock snapshot (issue #129) so a + // concurrent `record_execution` can never make the serialized (and hence + // checksummed) statistics a torn pre/post-write mix. + let d = self.read_consistent(); + // Serialize `stats_degraded` ONLY when it is `true` (dynamic 8/9-field // count). A non-degraded level then serializes in the exact pre-#117 // 8-field form — byte-identical to a v2 statistics payload persisted // before this flag existed — so a `PriceLevelSnapshotPackage`'s SHA-256 // checksum, recomputed over the re-serialized bytes on - // `validate` / `from_snapshot_json`, still matches. A degraded level - // adds the field; `Deserialize` / `FromStr` default a missing flag to - // `false`, so both directions round-trip. - let degraded = self.stats_degraded.load(Ordering::Relaxed); + // `validate` / `from_snapshot_json`, still matches for BOTH a legacy v2 + // and a new v3 non-degraded package (issue #129 keeps checksum + // recomputation version-agnostic — see `SNAPSHOT_FORMAT_VERSION`). A + // degraded level adds the 9th field (the v3-only shape); `Deserialize` / + // `FromStr` default a missing flag to `false`, so both directions + // round-trip. + let degraded = d.stats_degraded; let field_count = if degraded { 9 } else { 8 }; let mut state = serializer.serialize_struct("PriceLevelStatistics", field_count)?; - state.serialize_field("orders_added", &self.orders_added.load(Ordering::Relaxed))?; - state.serialize_field( - "orders_removed", - &self.orders_removed.load(Ordering::Relaxed), - )?; - state.serialize_field( - "orders_executed", - &self.orders_executed.load(Ordering::Relaxed), - )?; - state.serialize_field( - "quantity_executed", - &self.quantity_executed.load(Ordering::Relaxed), - )?; - state.serialize_field( - "value_executed", - &self.value_executed.load(Ordering::Relaxed), - )?; - state.serialize_field( - "last_execution_time", - &self.last_execution_time.load(Ordering::Relaxed), - )?; - state.serialize_field( - "first_arrival_time", - &self.first_arrival_time.load(Ordering::Relaxed), - )?; - state.serialize_field( - "sum_waiting_time", - &self.sum_waiting_time.load(Ordering::Relaxed), - )?; + state.serialize_field("orders_added", &d.orders_added)?; + state.serialize_field("orders_removed", &d.orders_removed)?; + state.serialize_field("orders_executed", &d.orders_executed)?; + state.serialize_field("quantity_executed", &d.quantity_executed)?; + state.serialize_field("value_executed", &d.value_executed)?; + state.serialize_field("last_execution_time", &d.last_execution_time)?; + state.serialize_field("first_arrival_time", &d.first_arrival_time)?; + state.serialize_field("sum_waiting_time", &d.sum_waiting_time)?; if degraded { state.serialize_field("stats_degraded", &true)?; } @@ -785,6 +950,7 @@ impl<'de> Deserialize<'de> for PriceLevelStatistics { first_arrival_time: AtomicU64::new(first_arrival_time), sum_waiting_time: AtomicU64::new(sum_waiting_time), stats_degraded: AtomicBool::new(stats_degraded), + stats_seq: AtomicU64::new(0), }) } } diff --git a/src/price_level/tests/snapshot.rs b/src/price_level/tests/snapshot.rs index bdafeb5..8a7673d 100644 --- a/src/price_level/tests/snapshot.rs +++ b/src/price_level/tests/snapshot.rs @@ -200,6 +200,87 @@ mod tests { assert!(matches!(err, PriceLevelError::InvalidOperation { .. })); } + #[test] + fn test_snapshot_v3_roundtrips_degraded_and_non_degraded() { + // Issue #129: new packages are v3 and round-trip BOTH a non-degraded + // (8-field statistics) and a degraded (9-field statistics) payload. + use crate::price_level::PriceLevelStatistics; + + // Non-degraded. + let clean = PriceLevelStatistics::new(); + clean + .record_execution(5, 100, 0, 1_000) + .expect("clean record"); + assert!(!clean.stats_degraded()); + let snap = PriceLevelSnapshot::with_orders_and_stats( + Price::new(10), + create_sample_orders(), + clean, + ) + .expect("snapshot"); + let package = PriceLevelSnapshotPackage::new(snap).expect("package"); + assert_eq!(package.version(), SNAPSHOT_FORMAT_VERSION); + assert_eq!(package.version(), 3); + let json = package.to_json().expect("to_json"); + let restored = PriceLevelSnapshotPackage::from_json(&json) + .expect("from_json") + .into_snapshot() + .expect("v3 non-degraded must validate + restore"); + assert!(!restored.statistics().stats_degraded()); + + // Degraded: force a dropped execution (maker in the future of execution). + let degraded = PriceLevelStatistics::new(); + assert!(degraded.record_execution(1, 1, 5_000, 1_000).is_err()); + assert!(degraded.stats_degraded()); + let snap = PriceLevelSnapshot::with_orders_and_stats( + Price::new(10), + create_sample_orders(), + degraded, + ) + .expect("snapshot"); + let package = PriceLevelSnapshotPackage::new(snap).expect("package"); + assert_eq!(package.version(), 3); + let json = package.to_json().expect("to_json"); + assert!( + json.contains("stats_degraded"), + "a degraded v3 payload carries the 9th field" + ); + let restored = PriceLevelSnapshotPackage::from_json(&json) + .expect("from_json") + .into_snapshot() + .expect("v3 degraded must validate + restore"); + assert!( + restored.statistics().stats_degraded(), + "the degraded flag round-trips through a v3 snapshot" + ); + } + + #[test] + fn test_snapshot_v2_legacy_package_restores() { + // Issue #129: a legacy v2 package (8-field statistics, checksum over the + // v2 bytes) must still validate + restore — checksum recomputation is + // version-agnostic, and `validate` accepts both v2 and v3. + let snap = PriceLevelSnapshot::with_orders(Price::new(77), create_sample_orders()) + .expect("snapshot"); + let package = PriceLevelSnapshotPackage::new(snap).expect("package"); + // Relabel the (non-degraded, 8-field) payload as the legacy v2 version. + // The checksum is computed over the snapshot, not the version, so it + // stays valid — exactly the shape an old writer produced. + let json = package.to_json().expect("to_json"); + let mut value: Value = serde_json::from_str(&json).expect("parse"); + if let Some(obj) = value.as_object_mut() { + obj.insert("version".to_string(), Value::Number(2u32.into())); + } + let v2_json = serde_json::to_string(&value).expect("reserialize"); + + let v2_package = PriceLevelSnapshotPackage::from_json(&v2_json).expect("from_json"); + assert_eq!(v2_package.version(), 2); + let restored = v2_package + .into_snapshot() + .expect("legacy v2 package must validate + restore"); + assert!(!restored.statistics().stats_degraded()); + } + #[test] fn test_new() { let snapshot = PriceLevelSnapshot::new(Price::new(1000)); diff --git a/src/price_level/tests/statistics.rs b/src/price_level/tests/statistics.rs index 07f076b..3a89c82 100644 --- a/src/price_level/tests/statistics.rs +++ b/src/price_level/tests/statistics.rs @@ -598,4 +598,158 @@ mod tests { serde_json::from_str(°raded_json).expect("deserialize degraded"); assert!(degraded_back.stats_degraded(), "the flag must round-trip"); } + + #[test] + fn test_orders_executed_overflow_seeded_via_fromstr_is_all_or_nothing() { + // Issue #129: `orders_executed` can be seeded to `usize::MAX` through + // FromStr / serde. A later record's checked `+1` must reject the overflow + // all-or-nothing — no counter wraps, and no OTHER aggregate advances. + let seeded = format!( + "PriceLevelStatistics:orders_added=0;orders_removed=0;orders_executed={};quantity_executed=0;value_executed=0;last_execution_time=0;first_arrival_time=0;sum_waiting_time=0;stats_degraded=false", + usize::MAX + ); + let stats = PriceLevelStatistics::from_str(&seeded).expect("parse MAX orders_executed"); + assert_eq!(stats.orders_executed(), usize::MAX); + + assert!( + stats.record_execution(5, 100, 0, 1_000).is_err(), + "the checked orders_executed overflow must be rejected" + ); + // No wrap, and nothing else moved (all-or-nothing). + assert_eq!(stats.orders_executed(), usize::MAX, "no wrap to 0"); + assert_eq!(stats.quantity_executed(), 0); + assert_eq!(stats.value_executed(), 0); + assert_eq!(stats.sum_waiting_time(), 0); + assert!( + stats.stats_degraded(), + "the dropped execution is observable" + ); + } + + #[test] + fn test_mark_degraded_transitions_once() { + // Issue #129: only the FIRST drop transitions the sticky flag + // (compare_exchange), so `PriceLevel::match_order` logs one WARN per + // degraded episode rather than one per dropped execution. + let stats = PriceLevelStatistics::new(); + assert!(!stats.stats_degraded()); + assert!( + stats.mark_degraded(), + "first drop transitions false -> true" + ); + assert!(stats.stats_degraded()); + assert!(!stats.mark_degraded(), "second drop is silent"); + assert!(!stats.mark_degraded(), "still silent"); + // A reset re-arms the transition. + stats.reset(); + assert!(!stats.stats_degraded()); + assert!(stats.mark_degraded(), "post-reset drop transitions again"); + } + + #[test] + fn test_last_execution_time_is_monotonic() { + // Issue #129: `fetch_max` — recording an OLDER execution after a newer + // one never moves `last_execution_time` backwards. + let stats = PriceLevelStatistics::new(); + stats.record_execution(1, 100, 0, 5_000).expect("newer"); + assert_eq!(stats.last_execution_time(), 5_000); + stats.record_execution(1, 100, 0, 2_000).expect("older"); + assert_eq!( + stats.last_execution_time(), + 5_000, + "an older record must not move the last-execution time back" + ); + stats.record_execution(1, 100, 0, 9_000).expect("newest"); + assert_eq!(stats.last_execution_time(), 9_000); + } + + #[test] + fn test_last_execution_time_max_wins_under_barrier() { + // Issue #129: two records with distinct timestamps racing — the max wins + // regardless of order (`fetch_max` is atomic, independent of the seqlock). + use std::sync::{Arc as StdArc, Barrier}; + + const HI: u64 = 9_000; + const LO: u64 = 3_000; + for _ in 0..500 { + let stats = StdArc::new(PriceLevelStatistics::new()); + let barrier = StdArc::new(Barrier::new(2)); + let hi = { + let stats = StdArc::clone(&stats); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + let _ = stats.record_execution(1, 100, 0, HI); + }) + }; + let lo = { + let stats = StdArc::clone(&stats); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + let _ = stats.record_execution(1, 100, 0, LO); + }) + }; + hi.join().expect("hi"); + lo.join().expect("lo"); + assert_eq!( + stats.last_execution_time(), + HI, + "the max timestamp must win" + ); + } + } + + #[test] + fn test_clone_is_consistent_under_concurrent_record() { + // Issue #129 (F5): a `Clone` (which backs the checksummed snapshot) must + // capture a COHERENT set under a concurrent single writer. Each record + // adds qty=1 / value=100 / orders+1 inside the seqlock, so a consistent + // copy always has `value == 100 * quantity` AND `orders == quantity`. A + // pre-#129 torn clone could mix `orders=k` with `quantity=k-1`. + use std::sync::{Arc as StdArc, Barrier}; + + const N: u64 = 3_000; + let stats = StdArc::new(PriceLevelStatistics::new()); + let barrier = StdArc::new(Barrier::new(2)); + + let writer = { + let stats = StdArc::clone(&stats); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + for _ in 0..N { + stats.record_execution(1, 100, 0, 1_000).expect("record"); + } + }) + }; + let reader = { + let stats = StdArc::clone(&stats); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + for _ in 0..50_000 { + let copy = (*stats).clone(); + let q = copy.quantity_executed(); + assert_eq!( + copy.value_executed(), + q * 100, + "clone must not tear value vs quantity" + ); + assert_eq!( + copy.orders_executed() as u64, + q, + "clone must not tear orders vs quantity" + ); + } + }) + }; + + writer.join().expect("writer"); + reader.join().expect("reader"); + + assert_eq!(stats.orders_executed() as u64, N); + assert_eq!(stats.quantity_executed(), N); + assert_eq!(stats.value_executed(), N * 100); + } } From 059b81c7827897a085935142b4f17f6f0d8b2abb Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 15:21:59 +0200 Subject: [PATCH 3/4] fix(price_level): make PostOnly and FOK decisions atomic with the sweep PostOnly's pre-check and FOK's dry-run inspected the queue and then ran a separate sweep, so concurrent add_order / cancel / update could change matchable depth between the phases: a PostOnly taker could trade against depth added after its check, and an FOK taker could start with sufficient depth and return partially filled. PostOnly is now structural: a positive post-only either crosses (rejected) or rests (NotFilled), returning before any queue touch -- it never enters the sweep, so "PostOnly emits zero trades" holds under every interleaving with no lock. The old fall-through additionally let a non-crossing probe garbage-collect an unmatchable zero-visible reserve; a read-only probe no longer mutates resting liquidity. The reject-vs- rest decision is linearized at the depth-check read, and post-only is evaluated before TimeInForce, so a post-only taker never takes liquidity even under Fok/Ioc. FOK all-or-nothing is a cross-maker transactional property that per-entry atomics cannot express, so a positive FOK takes the exclusive side of a level RwLock across its feasibility check and sweep while add_order / update_order take the shared side; the Gtc/Ioc/Day sweep and cancel-through-update take no new ordering risk (guard always acquired before any entry lock; the sweep never re-enters a mutator). The guard is acquired exactly once per public update_order call -- the same-price Replace / UpdatePriceAndQuantity branches delegate to a guard-free inner function, so no recursive read can deadlock behind a queued FOK writer (reproduced pre-fix; the Barrier race test fails deterministically via a bounded timeout instead of hanging). Poisoned guards recover via into_inner. Lock-free headline claims in lib.rs/mod.rs/level.rs/README are qualified accordingly. Benches: uncontended shared acquisition on add/cancel within noise of the untouched Gtc control; post-only slightly faster (skips the vacuous sweep). Race tests (2000 iters each): add-vs-PostOnly never trades; cancel-vs-FOK and same-price-replace-vs-FOK are complete-or-killed, never partial, counters == queue every iteration. Closes #112 --- README.md | 21 +- src/lib.rs | 21 +- src/price_level/level.rs | 207 ++++++++++++++--- src/price_level/mod.rs | 15 +- src/price_level/tests/level.rs | 407 ++++++++++++++++++++++++++++++++- 5 files changed, 603 insertions(+), 68 deletions(-) diff --git a/README.md b/README.md index 79a1b35..8b5b410 100644 --- a/README.md +++ b/README.md @@ -12,13 +12,13 @@ # PriceLevel - A high-performance, lock-free price level implementation for limit order books in Rust. This library provides the building blocks for creating efficient trading systems with support for multiple order types and concurrent access patterns. + A high-performance price level implementation for limit order books in Rust, lock-free on its common add / match / cancel paths (a fill-or-kill match takes a brief level-exclusive guard — see below). This library provides the building blocks for creating efficient trading systems with support for multiple order types and concurrent access patterns. ## Features - - Lock-free architecture for high-throughput trading applications + - Lock-free common paths (add / match / cancel) for high-throughput trading applications; a fill-or-kill match briefly takes a level-exclusive guard and admissions / updates pay one uncontended shared acquisition (issue #112) - Support for diverse order types including standard limit orders, iceberg orders, post-only, fill-or-kill, and more - - Thread-safe operations with atomic counters and lock-free data structures + - Thread-safe operations built on atomic counters and lock-free data structures (with the fill-or-kill guard noted above) - Efficient order matching and execution logic - Designed with domain-driven principles for financial markets - Comprehensive test suite demonstrating concurrent usage scenarios @@ -52,7 +52,7 @@ ## Implementation Details - - **Thread Safety**: Uses atomic operations and lock-free data structures to ensure thread safety without mutex locks + - **Thread Safety**: Uses atomic operations and lock-free data structures — no locks on the common add / match / cancel paths; a fill-or-kill match briefly takes a level-exclusive guard (issue #112), and admissions / updates pay one uncontended shared acquisition - **Order Queue Management**: Specialized order queue keeping strict price-time priority via a lock-free `crossbeam-skiplist` ordered index - **Statistics Tracking**: Each price level tracks execution statistics in real-time - **Snapshot Capabilities**: Create point-in-time snapshots of price levels for market data distribution @@ -139,7 +139,7 @@ The simulation demonstrates the library's exceptional performance capabilities: - **High-Frequency Trading**: Over **264,000 operations per second** in realistic mixed workloads - **Hot Spot Performance**: Up to **7.75 million operations per second** under optimal conditions - **Write-Heavy Workloads**: Over **6.3 million operations per second** for pure write operations -- **Lock-Free Architecture**: Maintains high throughput with minimal contention overhead +- **Lock-Free Common Paths**: Maintains high throughput with minimal contention overhead on add / match / cancel (a fill-or-kill match takes a brief level-exclusive guard) The performance characteristics demonstrate that the `pricelevel` library is suitable for production use in high-performance trading systems, matching engines, and other financial applications where microsecond-level performance is critical. @@ -162,10 +162,13 @@ The performance characteristics demonstrate that the `pricelevel` library is sui are affected; nothing else is. - **Matching concurrency contract.** [`PriceLevel::match_order`] assumes a single logical matcher per level at a time. Concurrent `add_order` / - `cancel` from other threads are safe, but two concurrent `match_order` - calls on the *same* level — or a `match_order` racing a `cancel` of the - resting order it is currently matching — are the caller's responsibility - to serialize. + `update_order` (including a `cancel` of the resting order the matcher is + currently consuming) from other threads are safe and linearizable — the + match and the cancel serialize on the maker's per-entry lock (issue #81), + and a fill-or-kill match additionally takes a level-exclusive guard so it + stays all-or-nothing against those mutators (issue #112). Only two + concurrent `match_order` calls on the *same* level remain the caller's + responsibility to serialize. - **Reserve replenish amounts are now `NonZeroU64`** (issue #70). A replenish amount of `0` is structurally invalid: it would draw an empty visible tranche from the hidden quantity, silently leaving nothing diff --git a/src/lib.rs b/src/lib.rs index f2e16f5..04ea672 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,13 +4,13 @@ //! # PriceLevel //! -//! A high-performance, lock-free price level implementation for limit order books in Rust. This library provides the building blocks for creating efficient trading systems with support for multiple order types and concurrent access patterns. +//! A high-performance price level implementation for limit order books in Rust, lock-free on its common add / match / cancel paths (a fill-or-kill match takes a brief level-exclusive guard — see below). This library provides the building blocks for creating efficient trading systems with support for multiple order types and concurrent access patterns. //! //! ## Features //! -//! - Lock-free architecture for high-throughput trading applications +//! - Lock-free common paths (add / match / cancel) for high-throughput trading applications; a fill-or-kill match briefly takes a level-exclusive guard and admissions / updates pay one uncontended shared acquisition (issue #112) //! - Support for diverse order types including standard limit orders, iceberg orders, post-only, fill-or-kill, and more -//! - Thread-safe operations with atomic counters and lock-free data structures +//! - Thread-safe operations built on atomic counters and lock-free data structures (with the fill-or-kill guard noted above) //! - Efficient order matching and execution logic //! - Designed with domain-driven principles for financial markets //! - Comprehensive test suite demonstrating concurrent usage scenarios @@ -44,7 +44,7 @@ //! //! ## Implementation Details //! -//! - **Thread Safety**: Uses atomic operations and lock-free data structures to ensure thread safety without mutex locks +//! - **Thread Safety**: Uses atomic operations and lock-free data structures — no locks on the common add / match / cancel paths; a fill-or-kill match briefly takes a level-exclusive guard (issue #112), and admissions / updates pay one uncontended shared acquisition //! - **Order Queue Management**: Specialized order queue keeping strict price-time priority via a lock-free `crossbeam-skiplist` ordered index //! - **Statistics Tracking**: Each price level tracks execution statistics in real-time //! - **Snapshot Capabilities**: Create point-in-time snapshots of price levels for market data distribution @@ -131,7 +131,7 @@ //! - **High-Frequency Trading**: Over **264,000 operations per second** in realistic mixed workloads //! - **Hot Spot Performance**: Up to **7.75 million operations per second** under optimal conditions //! - **Write-Heavy Workloads**: Over **6.3 million operations per second** for pure write operations -//! - **Lock-Free Architecture**: Maintains high throughput with minimal contention overhead +//! - **Lock-Free Common Paths**: Maintains high throughput with minimal contention overhead on add / match / cancel (a fill-or-kill match takes a brief level-exclusive guard) //! //! The performance characteristics demonstrate that the `pricelevel` library is suitable for production use in high-performance trading systems, matching engines, and other financial applications where microsecond-level performance is critical. //! @@ -154,10 +154,13 @@ //! are affected; nothing else is. //! - **Matching concurrency contract.** [`PriceLevel::match_order`] assumes a //! single logical matcher per level at a time. Concurrent `add_order` / -//! `cancel` from other threads are safe, but two concurrent `match_order` -//! calls on the *same* level — or a `match_order` racing a `cancel` of the -//! resting order it is currently matching — are the caller's responsibility -//! to serialize. +//! `update_order` (including a `cancel` of the resting order the matcher is +//! currently consuming) from other threads are safe and linearizable — the +//! match and the cancel serialize on the maker's per-entry lock (issue #81), +//! and a fill-or-kill match additionally takes a level-exclusive guard so it +//! stays all-or-nothing against those mutators (issue #112). Only two +//! concurrent `match_order` calls on the *same* level remain the caller's +//! responsibility to serialize. //! - **Reserve replenish amounts are now `NonZeroU64`** (issue #70). A //! replenish amount of `0` is structurally invalid: it would draw an empty //! visible tranche from the hidden quantity, silently leaving nothing diff --git a/src/price_level/level.rs b/src/price_level/level.rs index e7327b0..383e2f0 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -11,8 +11,8 @@ use serde::{Deserialize, Serialize}; use std::fmt::Display; use std::str::FromStr; -use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, RwLock}; /// Bit layout of the [`PriceLevel::topology`] word (issue #126): the high two /// bits carry the pinned-side tag, the low bits the resting-order count. Packing @@ -64,7 +64,14 @@ mod topology { } } -/// A lock-free implementation of a price level in a limit order book. +/// A price level in a limit order book, lock-free on its common paths. +/// +/// Add, match (`Gtc` / `Ioc` / `Day`), and cancel take no lock — they run on +/// atomic counters and lock-free structures. A fill-or-kill match is the one +/// exception: it takes a level-exclusive guard across its feasibility check and +/// sweep so it stays all-or-nothing against concurrent mutation, and admissions +/// / updates pay one uncontended shared acquisition (see the `fok_guard` field +/// and [`Self::match_order`] for the full argument — issue #112). /// /// # Topology /// @@ -128,6 +135,26 @@ pub struct PriceLevel { /// Statistics for this price level stats: Arc, + + /// Fill-or-kill exclusion guard (issue #112). + /// + /// A fill-or-kill match must be **all-or-nothing** with respect to + /// concurrent `add_order` / `update_order` (cancel / resize): between its + /// depth dry-run and its sweep, a concurrent shrink could otherwise leave it + /// partially filled. All-or-nothing across N makers is a cross-maker + /// transactional property the per-entry atomics cannot express, so the FOK + /// path takes this guard's **write** (exclusive) side across its dry-run and + /// sweep, and the mutators ([`Self::add_order`] / [`Self::update_order`]) + /// take the **read** (shared) side. The common paths therefore pay only an + /// uncontended shared acquisition; a FOK (a cold, specific TIF) excludes + /// mutators for the duration of its feasibility check and sweep — an + /// `O(depth)` exclusive section, not a constant-time one. The non-FOK sweep + /// takes NO guard — it relies on the + /// single-matcher-per-level model and the existing per-entry cancel + /// atomicity (issue #81). The guarded value is `()`; a poisoned lock (a + /// holder panicked) is recovered with `into_inner`, since there is no state + /// to corrupt. + fok_guard: RwLock<()>, } impl PriceLevel { @@ -221,6 +248,7 @@ impl PriceLevel { topology_epoch: AtomicU64::new(0), orders: queue, stats: Arc::new(stats), + fok_guard: RwLock::new(()), }) } @@ -274,6 +302,7 @@ impl PriceLevel { topology_epoch: AtomicU64::new(0), orders: OrderQueue::new(), stats: Arc::new(PriceLevelStatistics::new()), + fok_guard: RwLock::new(()), } } @@ -483,6 +512,27 @@ impl PriceLevel { self.stats.clone() } + /// Acquire the fill-or-kill guard's **shared (read)** side — the mutator + /// side. Multiple mutators proceed concurrently; a fill-or-kill match + /// (holding the exclusive side) excludes them. A poisoned lock is recovered + /// with `into_inner` because the guarded value is `()` (see the field doc). + #[inline] + fn fok_read(&self) -> std::sync::RwLockReadGuard<'_, ()> { + self.fok_guard + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Acquire the fill-or-kill guard's **exclusive (write)** side — held across + /// a fill-or-kill dry-run + sweep so no mutator can change the matchable + /// depth mid-decision. A poisoned lock is recovered (see [`Self::fok_read`]). + #[inline] + fn fok_write(&self) -> std::sync::RwLockWriteGuard<'_, ()> { + self.fok_guard + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + /// Add an order to this price level. /// /// Decides the order's id IDENTITY first, then reserves its visible / @@ -542,6 +592,11 @@ impl PriceLevel { /// already rests at this level. A duplicate id takes precedence over a /// counter overflow. In every case the level is unchanged. pub fn add_order(&self, order: OrderType<()>) -> Result>, PriceLevelError> { + // Hold the fill-or-kill guard's shared side for this admission so a + // concurrent fill-or-kill match sees a stable depth (issue #112). This + // is an uncontended shared acquisition in the common case (no FOK). + let _fok = self.fok_read(); + // -------- Admission topology invariants (cheapest checks, no mutation) -------- // // A level holds orders at exactly one price and one side. Reject a @@ -947,16 +1002,45 @@ impl PriceLevel { /// This method still assumes a **single logical matcher per level at a /// time**: two concurrent `match_order` calls on the *same* level are NOT /// made safe here and must be serialized by the caller (an order book - /// typically matches a level from a single thread). The post-only / - /// fill-or-kill pre-checks read the queue before the sweep; under that - /// single-matcher assumption no concurrent `match_order` can change the - /// matchable depth between the pre-check and the sweep. Concurrent - /// `add_order` from other threads is likewise safe **for counter / queue - /// integrity**. The single-side topology invariant carries an additional - /// requirement beyond that integrity: it holds only when a given level's - /// admissions arrive from one logical path (see the type-level note on - /// [`PriceLevel`]), because the side is derived from the live queue rather - /// than stored. + /// typically matches a level from a single thread). Concurrent `add_order` + /// from other threads is safe **for counter / queue integrity**. The + /// single-side topology invariant carries an additional requirement beyond + /// that integrity: it holds only when a given level's admissions arrive from + /// one logical path (see the type-level note on [`PriceLevel`]), because the + /// side is derived from the live queue rather than stored. + /// + /// ## PostOnly and fill-or-kill are atomic with the sweep (issue #112) + /// + /// Both the post-only and the fill-or-kill decisions are made + /// all-or-nothing with respect to concurrent `add_order` / `update_order`, + /// by two different mechanisms: + /// + /// * **PostOnly never enters the sweep.** A positive post-only taker either + /// crosses resting depth (rejected) or does not (rests) — in NEITHER case + /// is any resting order consumed, and in neither case does it sweep. + /// Because there is no sweep, no concurrently-added maker can be turned + /// into a trade: "post-only emits zero trades" is a *structural* property + /// that holds under every interleaving, needing no lock. The verdict is + /// linearized at the `has_matchable_depth` read — depth committed before + /// it is crossed (reject); depth committed after it is ordered behind this + /// taker (rest), the correct decision at the read instant. A non-crossing + /// post-only taker leaves the level completely untouched; the caller + /// re-admits the residual via `add_order`. The post-only decision is + /// evaluated *before* the [`TimeInForce`] branch, so a post-only taker + /// never takes liquidity even when its TIF is `Fok` or `Ioc` — the + /// never-cross kind dominates the fill-or-kill / immediate-or-cancel + /// semantics. + /// * **Fill-or-kill holds an exclusive guard across its dry-run and sweep.** + /// A positive fill-or-kill taker takes the write side of the level's + /// fill-or-kill guard (the `fok_guard` note on [`PriceLevel`]) before its + /// feasibility dry-run and holds it through the sweep, while `add_order` / + /// `update_order` take the shared side. No mutator can then change the + /// matchable depth between the dry-run and the sweep, so with a stable + /// queue the sweep consumes exactly what the dry-run predicted: a + /// fill-or-kill taker either fills in full or is killed with the queue and + /// counters untouched — never a partial fill. The guard is acquired only + /// for fill-or-kill; the ordinary (`Gtc` / `Ioc` / `Day`) sweep is + /// unguarded and pays nothing. /// /// # Statistics /// @@ -1007,28 +1091,43 @@ impl PriceLevel { // -------- Taker TIF / kind pre-checks (before any queue mutation) -------- // - // PostOnly: must never take liquidity. If any matchable depth exists for - // a positive taker, reject without touching the queue. A zero-quantity - // taker has nothing to cross, so it is not rejected (it falls through to - // the vacuous-complete sweep below). - if taker_kind.is_post_only() - && incoming_quantity > 0 - && self.has_matchable_depth(taker_order_id) - { - tracing::debug!( - taker_order_id = %taker_order_id, - incoming_quantity, - price = self.price, - "post-only taker rejected: would take liquidity" - ); - let mut result = MatchResult::new(taker_order_id, Quantity::new(incoming_quantity)); - result.mark_rejected(incoming_quantity); - return result; + // PostOnly: must NEVER take liquidity (issue #112). A positive PostOnly + // taker either crosses (reject) or does not (rest) — and in NEITHER case + // does it enter the sweep. Not entering the sweep is what makes "PostOnly + // emits zero trades" hold under *every* interleaving: no concurrent + // `add_order` can turn a resting decision into a trade, because there is + // no sweep to consume the newly-added depth. The verdict is linearized at + // the `has_matchable_depth` check: depth committed before it is crossed + // (reject); depth committed after it is ordered "later" (behind this + // taker), so resting is correct at the check instant. No guard is needed. + if taker_kind.is_post_only() && incoming_quantity > 0 { + if self.has_matchable_depth(taker_order_id) { + tracing::debug!( + taker_order_id = %taker_order_id, + incoming_quantity, + price = self.price, + "post-only taker rejected: would take liquidity" + ); + let mut result = MatchResult::new(taker_order_id, Quantity::new(incoming_quantity)); + result.mark_rejected(incoming_quantity); + return result; + } + // Not crossable: rest (report no trade) WITHOUT sweeping. The caller + // re-admits the residual via `add_order`. + return MatchResult::new(taker_order_id, Quantity::new(incoming_quantity)); } - // Fill-or-kill: all-or-nothing. If the level cannot fill the taker in - // full, kill it without touching the queue. - if matches!(taker_tif, TimeInForce::Fok) && incoming_quantity > 0 { + // Fill-or-kill: all-or-nothing w.r.t. concurrent mutators (issue #112). + // Take the fill-or-kill guard's EXCLUSIVE side and hold it across BOTH + // the feasibility dry-run and the sweep below, so no `add_order` / + // `update_order` can change the matchable depth between the two: with a + // stable queue the sweep consumes exactly what the dry-run predicted, so + // an FOK either fills in full or (insufficient depth) is killed with the + // queue and counters untouched — never a partial fill. `_fok_guard` is + // `Some` only for a positive FOK taker; it drops at the end of the + // method (after the sweep). The non-FOK paths take no guard. + let _fok_guard = if matches!(taker_tif, TimeInForce::Fok) && incoming_quantity > 0 { + let guard = self.fok_write(); let available = self.matchable_quantity(incoming_quantity, taker_order_id); if available < incoming_quantity { tracing::debug!( @@ -1042,7 +1141,10 @@ impl PriceLevel { result.mark_killed(incoming_quantity); return result; } - } + Some(guard) + } else { + None + }; // A single sweep emits at most one trade and at most one filled-order // id per resting order it actually consumes. Two independent upper @@ -1626,6 +1728,31 @@ impl PriceLevel { pub fn update_order( &self, update: OrderUpdate, + ) -> Result>>, PriceLevelError> { + // Hold the fill-or-kill guard's shared side for the whole update so a + // concurrent fill-or-kill match cannot observe the depth shrink (cancel + // / down-size) or grow mid-decision (issue #112). Uncontended in the + // common case (no FOK running). Acquired exactly ONCE here: the + // same-price branches of `UpdatePriceAndQuantity` / `Replace` delegate + // to the guard-free `update_order_inner` rather than recursing into this + // method, because std `RwLock` reads are NOT reentrant. A recursive + // `read()` with a writer queued between the two acquisitions (the lock + // is writer-preferring, so the queued writer blocks the second reader) + // would deadlock against a `fok_write` waiting on the first reader. + let _fok = self.fok_read(); + self.update_order_inner(update) + } + + /// Guard-free body of [`Self::update_order`]. + /// + /// The caller MUST already hold the fill-or-kill shared guard + /// ([`Self::fok_read`]); this method never acquires it. That is what lets + /// the same-price `UpdatePriceAndQuantity` / `Replace` branches re-enter it + /// without taking a second, non-reentrant [`std::sync::RwLock`] read + /// (issue #112). + fn update_order_inner( + &self, + update: OrderUpdate, ) -> Result>>, PriceLevelError> { match update { OrderUpdate::UpdatePrice { @@ -1811,8 +1938,11 @@ impl PriceLevel { } Ok(order) } else { - // If price is the same, just update the quantity (reuse logic) - self.update_order(OrderUpdate::UpdateQuantity { + // If price is the same, just update the quantity (reuse + // logic). Call the guard-free inner body — we already hold + // the fill-or-kill shared guard, and a second `fok_read` + // here would be a non-reentrant recursive read (issue #112). + self.update_order_inner(OrderUpdate::UpdateQuantity { order_id, new_quantity, }) @@ -1883,8 +2013,11 @@ impl PriceLevel { Ok(order) } else { - // If price is the same, just update the quantity - self.update_order(OrderUpdate::UpdateQuantity { + // If price is the same, just update the quantity. Call the + // guard-free inner body — we already hold the fill-or-kill + // shared guard, and a second `fok_read` here would be a + // non-reentrant recursive read (issue #112). + self.update_order_inner(OrderUpdate::UpdateQuantity { order_id, new_quantity: quantity, }) diff --git a/src/price_level/mod.rs b/src/price_level/mod.rs index b2b751a..e779c84 100644 --- a/src/price_level/mod.rs +++ b/src/price_level/mod.rs @@ -4,16 +4,21 @@ Date: 28/3/25 ******************************************************************************/ -//! Core price level module: lock-free order queue, matching, snapshots, and statistics. +//! Core price level module: order queue, matching, snapshots, and statistics, +//! lock-free on the common paths. //! //! This module provides the central [`PriceLevel`] type that represents a single price -//! point in a limit order book. It manages a lock-free queue of orders, performs matching, -//! tracks statistics, and supports snapshot persistence with checksum protection. +//! point in a limit order book. It manages a queue of orders (lock-free on add / match / +//! cancel), performs matching, tracks statistics, and supports snapshot persistence with +//! checksum protection. A fill-or-kill match takes a level-exclusive guard across its +//! feasibility check and sweep so it stays all-or-nothing against concurrent mutation +//! (issue #112). //! //! # Key Types //! -//! - [`PriceLevel`] — the main lock-free price level implementation supporting concurrent -//! add, match, update, and cancel operations via atomic counters and crossbeam queues. +//! - [`PriceLevel`] — the main price level implementation supporting concurrent add, match, +//! update, and cancel operations via atomic counters and crossbeam queues, lock-free on +//! those common paths (a fill-or-kill match takes a brief level-exclusive guard). //! - [`PriceLevelData`] — a serializable representation for data transfer and storage. //! - [`PriceLevelSnapshot`] — a point-in-time snapshot of all orders at a price level. //! - [`PriceLevelSnapshotPackage`] — a checksum-protected wrapper around a snapshot for diff --git a/src/price_level/tests/level.rs b/src/price_level/tests/level.rs index 3ab2f41..cb50e38 100644 --- a/src/price_level/tests/level.rs +++ b/src/price_level/tests/level.rs @@ -5222,10 +5222,10 @@ mod tests { assert_eq!(fok_level.order_count(), 1); assert_eq!(fok_level.hidden_quantity(), 50); - // PostOnly on a fresh level: no matchable depth -> not rejected. It then - // falls through to the sweep as an ordinary (zero-taking) taker, which - // garbage-collects the unmatchable reserve with no trade and keeps the - // counters consistent with the queue. + // PostOnly on a fresh level: no matchable depth -> not rejected. As of + // issue #112 a PostOnly taker NEVER enters the sweep, so it reports no + // trade and leaves the level completely untouched (it does not + // garbage-collect the unmatchable reserve — that is not PostOnly's job). let po_level = PriceLevel::new(10000); let po_gen = new_trade_id_generator(); po_level @@ -5249,11 +5249,12 @@ mod tests { .as_u64(), 0 ); - // The non-matchable reserve is dropped by the fall-through sweep; the - // hidden counter is decremented in lockstep with the queue removal. - assert_eq!(po_level.order_count(), 0); + // PostOnly no longer sweeps (issue #112): the unmatchable reserve is + // left resting and the counters are unchanged — a PostOnly that does not + // cross rests without mutating the level at all. + assert_eq!(po_level.order_count(), 1); assert_eq!(po_level.visible_quantity(), 0); - assert_eq!(po_level.hidden_quantity(), 0); + assert_eq!(po_level.hidden_quantity(), 50); } #[test] @@ -8043,6 +8044,396 @@ mod tests { assert!(!restored.stats().stats_degraded()); assert_eq!(restored.order_count(), 1); } + + // --------------------------------------------------------------------- + // Issue #112 — PostOnly / FOK decisions are atomic with the sweep. + // --------------------------------------------------------------------- + + #[test] + fn test_post_only_never_trades_under_concurrent_add() { + // A PostOnly taker must emit ZERO trades under every interleaving with a + // concurrent `add_order` of crossable depth. PostOnly never enters the + // sweep (issue #112): whether the added maker lands before the + // `has_matchable_depth` check (taker rejected) or after it (taker rests), + // there is no sweep to consume that depth — so no add can turn a + // no-trade decision into a fill. The maker always survives fully resting. + use std::sync::{Arc, Barrier}; + use std::thread; + + const ITERATIONS: usize = 2_000; + const PRICE: u128 = 10_000; + const MAKER_QTY: u64 = 50; + + for iter in 0..ITERATIONS { + let level = Arc::new(PriceLevel::new(PRICE)); + let barrier = Arc::new(Barrier::new(2)); + let maker_id_u64 = (iter as u64) * 2 + 1; + let maker_id = Id::from_u64(maker_id_u64); + let generator = Arc::new(UuidGenerator::new(Uuid::from_u128( + 0xB0B0_0000_0000_0000u128 + iter as u128, + ))); + + let adder = { + let level = Arc::clone(&level); + let barrier = Arc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + level + .add_order(OrderType::Standard { + id: maker_id, + price: Price::new(PRICE), + quantity: Quantity::new(MAKER_QTY), + side: Side::Sell, + user_id: Hash32::zero(), + timestamp: TimestampMs::new(1_600_000_000_000 + iter as u64), + time_in_force: TimeInForce::Gtc, + extra_fields: (), + }) + .expect("add_order should succeed"); + }) + }; + + let matcher = { + let level = Arc::clone(&level); + let barrier = Arc::clone(&barrier); + let generator = Arc::clone(&generator); + thread::spawn(move || { + barrier.wait(); + level.match_order( + MAKER_QTY, + Id::from_u64(maker_id_u64 + 1), + TimeInForce::Gtc, + TakerKind::PostOnly, + TimestampMs::new(1_700_000_000_000), + &generator, + ) + }) + }; + + adder.join().expect("adder thread panicked"); + let result = matcher.join().expect("matcher thread panicked"); + + // The load-bearing invariant: a PostOnly taker NEVER trades. + assert_eq!( + result.trades().len(), + 0, + "iter {iter}: PostOnly must never emit a trade" + ); + assert_eq!( + result + .executed_quantity() + .expect("executed_quantity") + .as_u64(), + 0, + "iter {iter}: PostOnly must never execute quantity" + ); + // Either it saw the maker (rejected) or it did not (rested), but it + // never consumed. The maker is left resting at full quantity. + let snapshot = level.snapshot(); + assert_eq!(snapshot.order_count(), 1, "iter {iter}: maker must survive"); + assert_eq!( + snapshot.visible_quantity().as_u64(), + MAKER_QTY, + "iter {iter}: PostOnly must not have consumed the maker" + ); + assert_counters_match_queue(&level); + } + } + + #[test] + fn test_fok_all_or_nothing_under_concurrent_cancel() { + // A fill-or-kill taker that races a cancel of one of the two makers it + // needs must be all-or-nothing (issue #112): it either fills in FULL + // (both makers consumed, two trades) or is KILLED with zero trades and + // the queue/counters untouched. It must NEVER partially fill. The FOK + // holds the level-wide write guard across both its feasibility dry-run + // and the sweep, so the cancel (a reader) is serialized entirely before + // or entirely after — the depth the dry-run sees is exactly what the + // sweep consumes. + use std::sync::{Arc, Barrier}; + use std::thread; + + const ITERATIONS: usize = 2_000; + const PRICE: u128 = 10_000; + const MAKER_QTY: u64 = 30; + const TAKER_QTY: u64 = 60; // needs BOTH makers + + for iter in 0..ITERATIONS { + let level = Arc::new(PriceLevel::new(PRICE)); + let id_a_u64 = (iter as u64) * 3 + 1; + let id_b_u64 = id_a_u64 + 1; + let id_a = Id::from_u64(id_a_u64); + let id_b = Id::from_u64(id_b_u64); + + level + .add_order(create_sell_standard_order(id_a_u64, PRICE, MAKER_QTY)) + .expect("add id_a"); + level + .add_order(create_sell_standard_order(id_b_u64, PRICE, MAKER_QTY)) + .expect("add id_b"); + + let barrier = Arc::new(Barrier::new(2)); + let generator = Arc::new(UuidGenerator::new(Uuid::from_u128( + 0xF0F0_0000_0000_0000u128 + iter as u128, + ))); + + let matcher = { + let level = Arc::clone(&level); + let barrier = Arc::clone(&barrier); + let generator = Arc::clone(&generator); + thread::spawn(move || { + barrier.wait(); + level.match_order( + TAKER_QTY, + Id::from_u64(id_b_u64 + 1), + TimeInForce::Fok, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ) + }) + }; + + let canceller = { + let level = Arc::clone(&level); + let barrier = Arc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + level + .update_order(OrderUpdate::Cancel { order_id: id_b }) + .expect("cancel must not error") + }) + }; + + let result = matcher.join().expect("matcher thread panicked"); + let cancelled = canceller.join().expect("canceller thread panicked"); + + if result.is_complete() { + // Full fill: both makers consumed, exactly two trades, nothing + // left over. The cancel of id_b then found it already gone. + assert_eq!( + result.trades().len(), + 2, + "iter {iter}: a complete FOK consumes both makers" + ); + assert_eq!( + result + .executed_quantity() + .expect("executed_quantity") + .as_u64(), + TAKER_QTY, + "iter {iter}: complete FOK executes the full taker" + ); + assert_eq!(result.remaining_quantity().as_u64(), 0, "iter {iter}"); + assert!( + cancelled.is_none(), + "iter {iter}: id_b was consumed, so the cancel finds nothing" + ); + } else { + // Killed: the cancel won, dropping id_b below the needed depth. + // Zero trades, full taker remaining, id_a left untouched. + assert!( + result.was_killed(), + "iter {iter}: a FOK is complete or killed, NEVER partial" + ); + assert_eq!( + result.trades().len(), + 0, + "iter {iter}: a killed FOK emits zero trades" + ); + assert_eq!( + result.remaining_quantity().as_u64(), + TAKER_QTY, + "iter {iter}: a killed FOK leaves the whole taker unfilled" + ); + assert_eq!( + cancelled.map(|o| o.id()), + Some(id_b), + "iter {iter}: the cancel that won removed id_b" + ); + let snapshot = level.snapshot(); + assert_eq!( + snapshot.order_count(), + 1, + "iter {iter}: only id_a remains after a killed FOK" + ); + assert_eq!( + snapshot.visible_quantity().as_u64(), + MAKER_QTY, + "iter {iter}: id_a is untouched by a killed FOK" + ); + } + + // Belt and suspenders: a partial fill is NEVER a valid FOK outcome. + assert!( + !(!result.trades().is_empty() && !result.is_complete()), + "iter {iter}: FOK produced a partial fill" + ); + assert_counters_match_queue(&level); + // The (unused) id_a binding documents the surviving maker's identity. + let _ = id_a; + } + } + + #[test] + fn test_fok_vs_same_price_replace_no_deadlock() { + // Regression for the #112 recursive-read deadlock. A same-price `Replace` + // / `UpdatePriceAndQuantity` delegates to the guard-free + // `update_order_inner`, so it takes the fill-or-kill shared guard exactly + // ONCE. Before that fix it re-entered `update_order`, taking a SECOND + // `RwLock` read on the same thread; with a `fok_write` (the FOK matcher) + // queued between the two reads, the writer-preferring lock deadlocked + // (matcher waits on read#2 behind the queued writer; the writer waits on + // read#1). Here a FOK matcher races a same-price resize of one of its + // makers under a `Barrier`; the run must COMPLETE (bounded by a channel + // timeout — a regression would hang, so the timeout turns a hang into a + // deterministic failure) and the FOK must stay all-or-nothing. + use std::sync::mpsc; + use std::sync::{Arc, Barrier}; + use std::thread; + use std::time::Duration; + + const ITERATIONS: usize = 2_000; + const PRICE: u128 = 10_000; + const MAKER_QTY: u64 = 30; + const TAKER_QTY: u64 = 60; // needs BOTH makers at full size + const RESIZE_QTY: u64 = 5; // shrink id_b so a Replace-wins interleaving kills the FOK + // Generous relative to the ~ms of real work; only trips on a true hang. + let deadline = Duration::from_secs(30); + + for iter in 0..ITERATIONS { + let level = Arc::new(PriceLevel::new(PRICE)); + let id_a_u64 = (iter as u64) * 3 + 1; + let id_b_u64 = id_a_u64 + 1; + let id_b = Id::from_u64(id_b_u64); + + level + .add_order(create_sell_standard_order(id_a_u64, PRICE, MAKER_QTY)) + .expect("add id_a"); + level + .add_order(create_sell_standard_order(id_b_u64, PRICE, MAKER_QTY)) + .expect("add id_b"); + + let barrier = Arc::new(Barrier::new(2)); + let generator = Arc::new(UuidGenerator::new(Uuid::from_u128( + 0xDEAD_0000_0000_0000u128 + iter as u128, + ))); + let (mtx, mrx) = mpsc::channel(); + let (utx, urx) = mpsc::channel(); + + let matcher = { + let level = Arc::clone(&level); + let barrier = Arc::clone(&barrier); + let generator = Arc::clone(&generator); + thread::spawn(move || { + barrier.wait(); + let r = level.match_order( + TAKER_QTY, + Id::from_u64(id_b_u64 + 1), + TimeInForce::Fok, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ); + let _ = mtx.send(r); + }) + }; + + // Alternate the two recursive same-price branches across iterations. + let mutator = { + let level = Arc::clone(&level); + let barrier = Arc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + let update = if iter % 2 == 0 { + OrderUpdate::Replace { + order_id: id_b, + price: Price::new(PRICE), + quantity: Quantity::new(RESIZE_QTY), + side: Side::Sell, + } + } else { + OrderUpdate::UpdatePriceAndQuantity { + order_id: id_b, + new_price: Price::new(PRICE), + new_quantity: Quantity::new(RESIZE_QTY), + } + }; + let r = level.update_order(update); + let _ = utx.send(r); + }) + }; + + // Bounded wait: on the pre-fix deadlock neither send arrives and this + // fails deterministically instead of hanging the whole suite. + let result = mrx + .recv_timeout(deadline) + .unwrap_or_else(|_| panic!("iter {iter}: FOK matcher deadlocked (recursive read)")); + let updated = urx + .recv_timeout(deadline) + .unwrap_or_else(|_| { + panic!("iter {iter}: same-price update deadlocked (recursive read)") + }) + .expect("update_order must not error"); + + matcher.join().expect("matcher thread panicked"); + mutator.join().expect("mutator thread panicked"); + + if result.is_complete() { + // FOK won the guard: it consumed both makers at full size, so the + // later same-price update found id_b already gone (`Ok(None)`). + assert_eq!(result.trades().len(), 2, "iter {iter}: complete FOK"); + assert_eq!( + result + .executed_quantity() + .expect("executed_quantity") + .as_u64(), + TAKER_QTY, + "iter {iter}: complete FOK executes the full taker" + ); + assert_eq!(result.remaining_quantity().as_u64(), 0, "iter {iter}"); + assert!( + updated.is_none(), + "iter {iter}: id_b was consumed, so the same-price update finds nothing" + ); + } else { + // The resize won: id_b shrank below the needed depth, so the FOK + // was killed with zero trades and id_a + the shrunk id_b rest. + assert!( + result.was_killed(), + "iter {iter}: a FOK is complete or killed, NEVER partial" + ); + assert_eq!(result.trades().len(), 0, "iter {iter}: killed FOK"); + assert_eq!( + result.remaining_quantity().as_u64(), + TAKER_QTY, + "iter {iter}: a killed FOK leaves the whole taker unfilled" + ); + assert_eq!( + updated.map(|o| o.id()), + Some(id_b), + "iter {iter}: the resize that won returned the prior id_b order" + ); + let snapshot = level.snapshot(); + assert_eq!( + snapshot.order_count(), + 2, + "iter {iter}: both makers rest after a killed FOK" + ); + assert_eq!( + snapshot.visible_quantity().as_u64(), + MAKER_QTY + RESIZE_QTY, + "iter {iter}: id_a full + id_b shrunk" + ); + } + + assert!( + !(!result.trades().is_empty() && !result.is_complete()), + "iter {iter}: FOK produced a partial fill" + ); + assert_counters_match_queue(&level); + } + } } #[cfg(test)] From b0995c5c2e10c3e4ca15a735632803369ad511aa Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 19:07:06 +0200 Subject: [PATCH 4/4] address review: exact FOK feasibility, linearizable post-only, fail-fast poison - the guarded fill-or-kill dry run now simulates the sweep's counter evolution exactly (mutators are frozen by the write guard): pure consumes decrement the projected visible counter, replenishments apply their checked net delta, and a would-abort headroom overflow terminates the projection where the real sweep would stop -- so a FOK the sweep cannot complete is killed with zero trades instead of partially filling (reviewer's boundary construction is a test) - a mutation epoch (bumped by every committed admission / update) brackets the post-only depth scan with a stable-epoch retry, giving the rest-vs-reject verdict a linearization point - snapshot() takes the shared side of the FOK guard for its materialization, so a checksummed snapshot can never capture a multi-maker fill-or-kill mid-transaction - a poisoned guard no longer silently reopens the level: recovery trips a sticky poisoned flag (one ERROR log on transition); admissions and updates fail fast and matching refuses while set, snapshots stay available for diagnostics -- defense in depth, the production sweep has no unwind path - lock-free claims reworded honestly: the Gtc/Ioc/Day match is lock-free; admissions and updates are shared-lock mutators that can block behind an O(depth) fill-or-kill writer (README regenerated) - a cfg(test) decision-boundary seam makes the add-after-post-only- check race deterministic; the FOK cancel window is documented as structurally closed by the write guard --- README.md | 8 +- src/lib.rs | 8 +- src/price_level/level.rs | 317 ++++++++++++++++++++++++++++++--- src/price_level/mod.rs | 15 +- src/price_level/tests/level.rs | 140 +++++++++++++++ 5 files changed, 446 insertions(+), 42 deletions(-) diff --git a/README.md b/README.md index 8b5b410..07e3fb4 100644 --- a/README.md +++ b/README.md @@ -12,11 +12,11 @@ # PriceLevel - A high-performance price level implementation for limit order books in Rust, lock-free on its common add / match / cancel paths (a fill-or-kill match takes a brief level-exclusive guard — see below). This library provides the building blocks for creating efficient trading systems with support for multiple order types and concurrent access patterns. + A high-performance price level implementation for limit order books in Rust. The `Gtc` / `Ioc` / `Day` match path is lock-free (atomics + sharded / skiplist structures); admissions and updates (cancel / resize) take the shared side of a per-level guard — normally uncontended, but they can block behind an `O(depth)` fill-or-kill match that holds the exclusive side (see below). This library provides the building blocks for creating efficient trading systems with support for multiple order types and concurrent access patterns. ## Features - - Lock-free common paths (add / match / cancel) for high-throughput trading applications; a fill-or-kill match briefly takes a level-exclusive guard and admissions / updates pay one uncontended shared acquisition (issue #112) + - Lock-free `Gtc` / `Ioc` / `Day` match path for high-throughput trading; admissions and updates (cancel / resize) are shared-lock mutators — one normally-uncontended shared acquisition that can block behind an `O(depth)` fill-or-kill match holding the exclusive side (issue #112) - Support for diverse order types including standard limit orders, iceberg orders, post-only, fill-or-kill, and more - Thread-safe operations built on atomic counters and lock-free data structures (with the fill-or-kill guard noted above) - Efficient order matching and execution logic @@ -52,7 +52,7 @@ ## Implementation Details - - **Thread Safety**: Uses atomic operations and lock-free data structures — no locks on the common add / match / cancel paths; a fill-or-kill match briefly takes a level-exclusive guard (issue #112), and admissions / updates pay one uncontended shared acquisition + - **Thread Safety**: Uses atomic operations and lock-free data structures. The `Gtc` / `Ioc` / `Day` match path takes no lock; admissions and updates (cancel / resize) take the shared side of a per-level guard — normally uncontended, but they can block behind an `O(depth)` fill-or-kill match that holds the exclusive side (issue #112) - **Order Queue Management**: Specialized order queue keeping strict price-time priority via a lock-free `crossbeam-skiplist` ordered index - **Statistics Tracking**: Each price level tracks execution statistics in real-time - **Snapshot Capabilities**: Create point-in-time snapshots of price levels for market data distribution @@ -139,7 +139,7 @@ The simulation demonstrates the library's exceptional performance capabilities: - **High-Frequency Trading**: Over **264,000 operations per second** in realistic mixed workloads - **Hot Spot Performance**: Up to **7.75 million operations per second** under optimal conditions - **Write-Heavy Workloads**: Over **6.3 million operations per second** for pure write operations -- **Lock-Free Common Paths**: Maintains high throughput with minimal contention overhead on add / match / cancel (a fill-or-kill match takes a brief level-exclusive guard) +- **Lock-Free Match Path**: The `Gtc` / `Ioc` / `Day` match runs lock-free with minimal contention overhead; admissions / updates take a normally-uncontended shared lock that can block behind an `O(depth)` fill-or-kill match The performance characteristics demonstrate that the `pricelevel` library is suitable for production use in high-performance trading systems, matching engines, and other financial applications where microsecond-level performance is critical. diff --git a/src/lib.rs b/src/lib.rs index 04ea672..ed5d18e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,11 +4,11 @@ //! # PriceLevel //! -//! A high-performance price level implementation for limit order books in Rust, lock-free on its common add / match / cancel paths (a fill-or-kill match takes a brief level-exclusive guard — see below). This library provides the building blocks for creating efficient trading systems with support for multiple order types and concurrent access patterns. +//! A high-performance price level implementation for limit order books in Rust. The `Gtc` / `Ioc` / `Day` match path is lock-free (atomics + sharded / skiplist structures); admissions and updates (cancel / resize) take the shared side of a per-level guard — normally uncontended, but they can block behind an `O(depth)` fill-or-kill match that holds the exclusive side (see below). This library provides the building blocks for creating efficient trading systems with support for multiple order types and concurrent access patterns. //! //! ## Features //! -//! - Lock-free common paths (add / match / cancel) for high-throughput trading applications; a fill-or-kill match briefly takes a level-exclusive guard and admissions / updates pay one uncontended shared acquisition (issue #112) +//! - Lock-free `Gtc` / `Ioc` / `Day` match path for high-throughput trading; admissions and updates (cancel / resize) are shared-lock mutators — one normally-uncontended shared acquisition that can block behind an `O(depth)` fill-or-kill match holding the exclusive side (issue #112) //! - Support for diverse order types including standard limit orders, iceberg orders, post-only, fill-or-kill, and more //! - Thread-safe operations built on atomic counters and lock-free data structures (with the fill-or-kill guard noted above) //! - Efficient order matching and execution logic @@ -44,7 +44,7 @@ //! //! ## Implementation Details //! -//! - **Thread Safety**: Uses atomic operations and lock-free data structures — no locks on the common add / match / cancel paths; a fill-or-kill match briefly takes a level-exclusive guard (issue #112), and admissions / updates pay one uncontended shared acquisition +//! - **Thread Safety**: Uses atomic operations and lock-free data structures. The `Gtc` / `Ioc` / `Day` match path takes no lock; admissions and updates (cancel / resize) take the shared side of a per-level guard — normally uncontended, but they can block behind an `O(depth)` fill-or-kill match that holds the exclusive side (issue #112) //! - **Order Queue Management**: Specialized order queue keeping strict price-time priority via a lock-free `crossbeam-skiplist` ordered index //! - **Statistics Tracking**: Each price level tracks execution statistics in real-time //! - **Snapshot Capabilities**: Create point-in-time snapshots of price levels for market data distribution @@ -131,7 +131,7 @@ //! - **High-Frequency Trading**: Over **264,000 operations per second** in realistic mixed workloads //! - **Hot Spot Performance**: Up to **7.75 million operations per second** under optimal conditions //! - **Write-Heavy Workloads**: Over **6.3 million operations per second** for pure write operations -//! - **Lock-Free Common Paths**: Maintains high throughput with minimal contention overhead on add / match / cancel (a fill-or-kill match takes a brief level-exclusive guard) +//! - **Lock-Free Match Path**: The `Gtc` / `Ioc` / `Day` match runs lock-free with minimal contention overhead; admissions / updates take a normally-uncontended shared lock that can block behind an `O(depth)` fill-or-kill match //! //! The performance characteristics demonstrate that the `pricelevel` library is suitable for production use in high-performance trading systems, matching engines, and other financial applications where microsecond-level performance is critical. //! diff --git a/src/price_level/level.rs b/src/price_level/level.rs index 383e2f0..38559d2 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -11,7 +11,7 @@ use serde::{Deserialize, Serialize}; use std::fmt::Display; use std::str::FromStr; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, RwLock}; /// Bit layout of the [`PriceLevel::topology`] word (issue #126): the high two @@ -64,14 +64,67 @@ mod topology { } } -/// A price level in a limit order book, lock-free on its common paths. +// Deterministic race seam for the post-only decision boundary (issue #130). +// +// `match_order` fires `fire_post_only_decision_hook` BETWEEN the post-only depth +// decision and its commit, so a test can install a hook that mutates the level +// (e.g. `add_order`) in that exact window and assert the outcome +// deterministically, without relying on scheduler stress. Production builds +// compile none of this — the hook call site is `#[cfg(test)]`. +#[cfg(test)] +thread_local! { + static POST_ONLY_DECISION_HOOK: std::cell::RefCell>> = + const { std::cell::RefCell::new(None) }; +} + +/// Install a hook fired at the post-only decision boundary (test seam, issue +/// #130). Returns a guard that clears the hook on drop. +#[cfg(test)] +pub(crate) fn set_post_only_decision_hook(hook: Box) -> PostOnlyHookGuard { + POST_ONLY_DECISION_HOOK.with(|slot| *slot.borrow_mut() = Some(hook)); + PostOnlyHookGuard +} + +/// Clears the post-only decision hook when dropped (test seam, issue #130). +#[cfg(test)] +pub(crate) struct PostOnlyHookGuard; + +#[cfg(test)] +impl Drop for PostOnlyHookGuard { + fn drop(&mut self) { + POST_ONLY_DECISION_HOOK.with(|slot| *slot.borrow_mut() = None); + } +} + +/// Fire the post-only decision hook if one is installed (test seam, issue #130). +#[cfg(test)] +fn fire_post_only_decision_hook() { + // Take the hook OUT of the slot while firing so a re-entrant `match_order` + // inside the hook does not double-borrow the `RefCell`. + let hook = POST_ONLY_DECISION_HOOK.with(|slot| slot.borrow_mut().take()); + if let Some(mut hook) = hook { + hook(); + POST_ONLY_DECISION_HOOK.with(|slot| { + let mut slot = slot.borrow_mut(); + if slot.is_none() { + *slot = Some(hook); + } + }); + } +} + +/// A price level in a limit order book, lock-free on the match path. /// -/// Add, match (`Gtc` / `Ioc` / `Day`), and cancel take no lock — they run on -/// atomic counters and lock-free structures. A fill-or-kill match is the one -/// exception: it takes a level-exclusive guard across its feasibility check and -/// sweep so it stays all-or-nothing against concurrent mutation, and admissions -/// / updates pay one uncontended shared acquisition (see the `fok_guard` field -/// and [`Self::match_order`] for the full argument — issue #112). +/// A `Gtc` / `Ioc` / `Day` match runs entirely on atomic counters and lock-free +/// (sharded / skiplist) structures — no lock. The mutators [`Self::add_order`] +/// and [`Self::update_order`] (cancel and resize included) are NOT lock-free: +/// each takes the **shared** side of a per-level reader-writer guard. That +/// acquisition is normally uncontended (it only coordinates with a fill-or-kill +/// match), but it can BLOCK behind a concurrent fill-or-kill: a `Fok` match +/// takes the guard's **exclusive** side across its feasibility check and sweep — +/// an `O(depth)` critical section — so it stays all-or-nothing against +/// concurrent mutation. See the `fok_guard` field and [`Self::match_order`] for +/// the full argument (issue #112). /// /// # Topology /// @@ -151,10 +204,30 @@ pub struct PriceLevel { /// `O(depth)` exclusive section, not a constant-time one. The non-FOK sweep /// takes NO guard — it relies on the /// single-matcher-per-level model and the existing per-entry cancel - /// atomicity (issue #81). The guarded value is `()`; a poisoned lock (a - /// holder panicked) is recovered with `into_inner`, since there is no state - /// to corrupt. + /// atomicity (issue #81). The guarded value is `()`, so a poisoned lock is + /// recovered with `into_inner` — but a poison means a holder PANICKED + /// mid-operation, which may have left the level half-mutated, so the recovery + /// also trips [`Self::level_poisoned`] and the level then fails fast rather + /// than silently reopening (issue #130). fok_guard: RwLock<()>, + + /// Sticky fail-fast flag set when a poisoned [`Self::fok_guard`] is recovered + /// (issue #130): a guard holder panicked mid-operation, so the level may be + /// half-mutated and cannot be trusted. While set, `add_order` / `update_order` + /// return [`PriceLevelError::InvalidOperation`] and `match_order` refuses to + /// match (returns an empty result); `snapshot` stays allowed for diagnostics + /// / reconstruction. The production match sweep has no unwind path (audited), + /// so this is defense-in-depth; a poisoned level requires reconstruction from + /// a snapshot. Never cleared once set. + level_poisoned: AtomicBool, + + /// Monotonic counter bumped by every committing `add_order` / `update_order` + /// (cancel / resize) so the non-atomic post-only depth scan can linearize + /// (issue #130). [`Self::has_matchable_depth`] reads it, scans the queue, + /// re-reads it, and retries on change: a stable epoch across the scan means + /// no mutation committed during it, giving the post-only verdict a + /// linearization point instead of a torn read. + mutation_epoch: AtomicU64, } impl PriceLevel { @@ -249,6 +322,8 @@ impl PriceLevel { orders: queue, stats: Arc::new(stats), fok_guard: RwLock::new(()), + level_poisoned: AtomicBool::new(false), + mutation_epoch: AtomicU64::new(0), }) } @@ -303,6 +378,8 @@ impl PriceLevel { orders: OrderQueue::new(), stats: Arc::new(PriceLevelStatistics::new()), fok_guard: RwLock::new(()), + level_poisoned: AtomicBool::new(false), + mutation_epoch: AtomicU64::new(0), } } @@ -491,6 +568,14 @@ impl PriceLevel { self.topology_epoch.fetch_add(1, Ordering::Release); } + /// Bump the mutation epoch on a committed add / cancel / resize so a racing + /// post-only depth scan retries (issue #130). `Release` so the queue mutation + /// that precedes it happens-before a scanner's `Acquire` read of the epoch. + #[inline] + fn bump_mutation_epoch(&self) { + self.mutation_epoch.fetch_add(1, Ordering::Release); + } + /// Returns `true` if `orders` is empty or every order shares one side — the /// single-side coherence [`Self::from_snapshot`] requires. Used as the /// termination backstop for `snapshot`'s torn-topology retry (issue #126). @@ -515,12 +600,15 @@ impl PriceLevel { /// Acquire the fill-or-kill guard's **shared (read)** side — the mutator /// side. Multiple mutators proceed concurrently; a fill-or-kill match /// (holding the exclusive side) excludes them. A poisoned lock is recovered - /// with `into_inner` because the guarded value is `()` (see the field doc). + /// with `into_inner` (the guarded value is `()`), but the recovery trips the + /// sticky [`Self::level_poisoned`] flag so the level then fails fast (issue + /// #130): a poison means a holder panicked mid-operation. #[inline] fn fok_read(&self) -> std::sync::RwLockReadGuard<'_, ()> { - self.fok_guard - .read() - .unwrap_or_else(std::sync::PoisonError::into_inner) + self.fok_guard.read().unwrap_or_else(|poison| { + self.mark_poisoned(); + poison.into_inner() + }) } /// Acquire the fill-or-kill guard's **exclusive (write)** side — held across @@ -528,9 +616,62 @@ impl PriceLevel { /// depth mid-decision. A poisoned lock is recovered (see [`Self::fok_read`]). #[inline] fn fok_write(&self) -> std::sync::RwLockWriteGuard<'_, ()> { - self.fok_guard - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner) + self.fok_guard.write().unwrap_or_else(|poison| { + self.mark_poisoned(); + poison.into_inner() + }) + } + + /// Trip the sticky poison flag when a [`Self::fok_guard`] poison is recovered + /// (issue #130). Logs `ERROR` exactly once — on the `false -> true` + /// transition decided by the `compare_exchange` — so a poisoned level is + /// reported but not flooded. + #[cold] + fn mark_poisoned(&self) { + if self + .level_poisoned + .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + tracing::error!( + price = self.price, + "price level poisoned by a panicked operation; matching and mutation are now refused — reconstruct the level from a snapshot" + ); + } + } + + /// Returns `true` if the level has been poisoned by a panicked guard holder + /// (issue #130). + #[inline] + #[must_use] + fn is_poisoned(&self) -> bool { + self.level_poisoned.load(Ordering::Relaxed) + } + + /// Fail-fast guard for the mutating public methods: `Err` once the level is + /// poisoned (issue #130). + #[inline] + fn poison_check(&self) -> Result<(), PriceLevelError> { + if self.is_poisoned() { + Err(PriceLevelError::InvalidOperation { + message: "price level poisoned by a panicked operation".to_string(), + }) + } else { + Ok(()) + } + } + + /// Genuinely poison the fill-or-kill guard by panicking while holding its + /// write side (issue #130 test seam). The panic is caught so the test + /// process survives; the `RwLock` is left poisoned, so the NEXT guard + /// acquisition recovers it and trips [`Self::level_poisoned`], exercising the + /// real fail-fast path (not a directly-set flag). + #[cfg(test)] + pub(crate) fn test_poison_guard(&self) { + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _guard = self.fok_write(); + panic!("intentional guard poison for test"); + })); } /// Add an order to this price level. @@ -596,6 +737,9 @@ impl PriceLevel { // concurrent fill-or-kill match sees a stable depth (issue #112). This // is an uncontended shared acquisition in the common case (no FOK). let _fok = self.fok_read(); + // Fail fast if a prior panic poisoned the guard (or this very acquisition + // just recovered one): the level may be half-mutated (issue #130). + self.poison_check()?; // -------- Admission topology invariants (cheapest checks, no mutation) -------- // @@ -736,6 +880,10 @@ impl PriceLevel { // Update statistics only after a committed admission. self.stats.record_order_added(); + // Signal the committed mutation so a racing post-only depth scan retries + // (issue #130). + self.bump_mutation_epoch(); + Ok(order_arc) } @@ -807,8 +955,25 @@ impl PriceLevel { /// self-trade prevention, so it is not liquidity this taker could take, and /// the post-only pre-check must agree. fn has_matchable_depth(&self, taker_id: Id) -> bool { - self.iter_orders() - .any(|order| order.id() != taker_id && order.is_matchable()) + // Linearize the non-atomic scan against concurrent mutation (issue #130): + // read the mutation epoch, scan, re-read; retry if it moved. A stable + // epoch across the scan means no add / cancel / resize committed during + // it, so the verdict corresponds to a single queue state and has a + // linearization point (at the scan). Unbounded retry, same liveness + // argument as the statistics seqlock: mutators are finite and each commit + // is a single `fetch_add`, so a scan converges as soon as mutation + // quiesces (the post-only path is cold, so the retry cost is irrelevant). + loop { + let epoch_before = self.mutation_epoch.load(Ordering::Acquire); + let verdict = self + .iter_orders() + .any(|order| order.id() != taker_id && order.is_matchable()); + std::sync::atomic::fence(Ordering::Acquire); + let epoch_after = self.mutation_epoch.load(Ordering::Relaxed); + if epoch_before == epoch_after { + return verdict; + } + } } /// Computes how much of `incoming_quantity` this level could actually fill @@ -819,9 +984,15 @@ impl PriceLevel { /// same price-time order the real sweep uses, including iceberg / reserve /// replenishment (a refreshed tranche is re-queued at the tail) and the /// removal of a non-replenishing reserve once its visible part is drained. - /// The returned value is therefore exactly what [`Self::match_order`] would - /// consume — never an over- or under-count — which is what fill-or-kill - /// (all-or-nothing) correctly depends on. + /// It also models the sweep's **replenish-headroom abort** (issue + /// #124/#130): it tracks the level's visible counter as the sweep would + /// evolve it and stops at the exact maker where a replenish's net delta would + /// overflow `u64`, because the real sweep sets that maker aside and + /// terminates. The returned value is therefore exactly what + /// [`Self::match_order`] would consume — never an over- or under-count — + /// which is what fill-or-kill (all-or-nothing) correctly depends on: without + /// modelling the abort, this could approve a fill-or-kill the sweep then + /// aborts mid-fill, leaving a partial fill. /// /// `taker_id` must be the id of the taker this depth is being computed for: /// a resting maker sharing that id is skipped, exactly as the real sweep @@ -854,6 +1025,19 @@ impl PriceLevel { let mut remaining = incoming_quantity; let mut filled: u64 = 0; + // Track the level's visible counter as the sweep would evolve it, so the + // dry run models the #124 replenish-headroom ABORT (issue #130). A + // replenish step commits `visible -= consumed; visible += hidden_reduced` + // with checked arithmetic under the entry lock, and if that net delta + // would exceed `u64::MAX` the real sweep SETS THE MAKER ASIDE (no trade) + // and TERMINATES the sweep. Starting from the live counter, this + // simulation reproduces that stop point exactly. When the fill-or-kill + // dry run holds the `fok_write` guard, mutators are frozen, so the live + // value cannot drift and the projection is exact — a would-abort here + // therefore correctly makes the fill-or-kill infeasible (killed) rather + // than approving a taker the sweep would abort mid-fill (a partial fill). + let mut projected_visible = self.visible_quantity(); + while remaining > 0 { let Some(order) = pending.pop_front() else { break; @@ -885,6 +1069,32 @@ impl PriceLevel { continue; } + // Evolve the projected visible counter exactly as the sweep will, and + // STOP on a would-abort so this maker (and everything behind it) + // contributes nothing — the same terminal state the real sweep + // reaches (issue #130). + if hidden_reduced > 0 { + // Replenish: checked net delta `- consumed + hidden_reduced`. A + // failure is the #124 abort: the maker is set aside untouched and + // the sweep ends, so do NOT count `consumed` and break. + match projected_visible + .checked_sub(consumed) + .and_then(|v| v.checked_add(hidden_reduced)) + { + Some(next) => projected_visible = next, + None => break, + } + } else { + // Pure consume: visible only decreases, so it cannot abort. Track + // it (checked, never wraps: `consumed <= projected_visible`) so a + // later replenish's headroom test is accurate; a defensive + // underflow stops the dry run conservatively. + let Some(next) = projected_visible.checked_sub(consumed) else { + break; + }; + projected_visible = next; + } + // `consumed <= remaining <= incoming_quantity`, so this sum cannot // overflow `u64`; checked anyway per the no-saturate/no-wrap rule. filled = match filled.checked_add(consumed) { @@ -1062,6 +1272,19 @@ impl PriceLevel { timestamp: TimestampMs, trade_id_generator: &UuidGenerator, ) -> MatchResult { + // -------- Fail-fast on a poisoned level (issue #130) -------- + // + // If a guard holder panicked mid-operation the level may be half-mutated + // and cannot be trusted to match. `match_order` returns [`MatchResult`], + // not `Result`, so it cannot surface `InvalidOperation` the way + // `add_order` / `update_order` do; instead it REFUSES to match — an empty + // result (no trades, full remaining) — which is the safe outcome (the + // taker takes no liquidity from a corrupt level). The one-time `ERROR` + // log was already emitted when the poison was first recovered. + if self.is_poisoned() { + return MatchResult::new(taker_order_id, Quantity::new(incoming_quantity)); + } + // -------- Self-match is terminal (issue #126, tightening #120) -------- // // If the taker's own id already rests at this level, the taker cannot @@ -1097,11 +1320,22 @@ impl PriceLevel { // emits zero trades" hold under *every* interleaving: no concurrent // `add_order` can turn a resting decision into a trade, because there is // no sweep to consume the newly-added depth. The verdict is linearized at - // the `has_matchable_depth` check: depth committed before it is crossed - // (reject); depth committed after it is ordered "later" (behind this - // taker), so resting is correct at the check instant. No guard is needed. + // the `has_matchable_depth` check, which re-scans under the mutation epoch + // so a concurrent add / cancel / resize racing the scan is DETECTED and + // the scan retried (issue #130) — a stable epoch across the scan gives the + // verdict a single-queue-state linearization point rather than a torn + // read. Depth committed before the linearized scan is crossed (reject); + // depth committed after it is ordered "later" (behind this taker), so + // resting is correct at the scan instant. No guard is needed. if taker_kind.is_post_only() && incoming_quantity > 0 { - if self.has_matchable_depth(taker_order_id) { + let crossable = self.has_matchable_depth(taker_order_id); + // Deterministic race seam (issue #130): fires BETWEEN the depth + // decision and the commit below so a test can inject an `add_order` + // in that exact window and confirm PostOnly still emits zero trades + // (there is no sweep to consume the added depth). No-op in production. + #[cfg(test)] + fire_post_only_decision_hook(); + if crossable { tracing::debug!( taker_order_id = %taker_order_id, incoming_quantity, @@ -1128,6 +1362,11 @@ impl PriceLevel { // method (after the sweep). The non-FOK paths take no guard. let _fok_guard = if matches!(taker_tif, TimeInForce::Fok) && incoming_quantity > 0 { let guard = self.fok_write(); + // Acquiring the write guard may have just recovered a poison; refuse + // to match a half-mutated level rather than sweep it (issue #130). + if self.is_poisoned() { + return MatchResult::new(taker_order_id, Quantity::new(incoming_quantity)); + } let available = self.matchable_quantity(incoming_quantity, taker_order_id); if available < incoming_quantity { tracing::debug!( @@ -1581,6 +1820,18 @@ impl PriceLevel { /// restore. #[must_use] pub fn snapshot(&self) -> PriceLevelSnapshot { + // Hold the fill-or-kill guard's SHARED side across the materialization + // (issue #130) so a snapshot can never capture a multi-maker fill-or-kill + // mid-transaction: the FOK holds the EXCLUSIVE side across its dry-run and + // sweep, so this read waits for it to fully commit or is excluded before + // it starts — the snapshot sees the pre- or post-FOK state, never a + // partial sweep. Ordinary mutators (`add_order` / `update_order`) also + // take the shared side, so they run concurrently with this read (read vs + // read) and are handled by the topology-epoch retry below for side + // transitions. `snapshot` intentionally does NOT poison-check: it stays + // available on a poisoned level for diagnostics / reconstruction. + let _fok = self.fok_read(); + // Materialize the orders exactly once, in queue-consumption (insertion // sequence) order so a snapshot round-trip re-enqueues them in identical // priority order; every aggregate is derived from this same snapshot so @@ -1740,7 +1991,17 @@ impl PriceLevel { // is writer-preferring, so the queued writer blocks the second reader) // would deadlock against a `fok_write` waiting on the first reader. let _fok = self.fok_read(); - self.update_order_inner(update) + // Fail fast on a poisoned level (issue #130). + self.poison_check()?; + let result = self.update_order_inner(update); + // A committed mutation (`Ok(Some(_))` — the order was found and + // cancelled / resized / moved) bumps the mutation epoch so a racing + // post-only depth scan retries (issue #130). `Ok(None)` (not found) and + // `Err` change nothing, so they do not bump. + if matches!(result, Ok(Some(_))) { + self.bump_mutation_epoch(); + } + result } /// Guard-free body of [`Self::update_order`]. diff --git a/src/price_level/mod.rs b/src/price_level/mod.rs index e779c84..08c4bea 100644 --- a/src/price_level/mod.rs +++ b/src/price_level/mod.rs @@ -5,20 +5,23 @@ ******************************************************************************/ //! Core price level module: order queue, matching, snapshots, and statistics, -//! lock-free on the common paths. +//! lock-free on the match path. //! //! This module provides the central [`PriceLevel`] type that represents a single price -//! point in a limit order book. It manages a queue of orders (lock-free on add / match / -//! cancel), performs matching, tracks statistics, and supports snapshot persistence with -//! checksum protection. A fill-or-kill match takes a level-exclusive guard across its +//! point in a limit order book. It manages a queue of orders, performs matching (the +//! `Gtc` / `Ioc` / `Day` match path is lock-free), tracks statistics, and supports +//! snapshot persistence with checksum protection. Admissions and updates (cancel / resize) +//! take the shared side of a per-level guard — normally uncontended, but they can block +//! behind an `O(depth)` fill-or-kill match that holds the exclusive side across its //! feasibility check and sweep so it stays all-or-nothing against concurrent mutation //! (issue #112). //! //! # Key Types //! //! - [`PriceLevel`] — the main price level implementation supporting concurrent add, match, -//! update, and cancel operations via atomic counters and crossbeam queues, lock-free on -//! those common paths (a fill-or-kill match takes a brief level-exclusive guard). +//! update, and cancel operations via atomic counters and crossbeam queues; the +//! `Gtc` / `Ioc` / `Day` match is lock-free, while admissions / updates take a +//! normally-uncontended shared lock that can block behind an `O(depth)` fill-or-kill. //! - [`PriceLevelData`] — a serializable representation for data transfer and storage. //! - [`PriceLevelSnapshot`] — a point-in-time snapshot of all orders at a price level. //! - [`PriceLevelSnapshotPackage`] — a checksum-protected wrapper around a snapshot for diff --git a/src/price_level/tests/level.rs b/src/price_level/tests/level.rs index cb50e38..d521cf2 100644 --- a/src/price_level/tests/level.rs +++ b/src/price_level/tests/level.rs @@ -8434,6 +8434,146 @@ mod tests { assert_counters_match_queue(&level); } } + + #[test] + fn test_fok_killed_on_replenish_headroom_abort() { + // Issue #130: the fill-or-kill dry run must model the sweep's + // replenish-headroom ABORT. Reviewer's construction — FIFO [standard 1, + // auto-reserve(visible 1, replenish 100), standard u64::MAX-2] with the + // level's visible counter at capacity (u64::MAX). A FOK(2) trades maker 1 + // and then the reserve would replenish (visible net delta +99) which + // overflows u64, so the real sweep ABORTS mid-fill. matchable_quantity + // must model that stop (returns 1, not 2), so the FOK is KILLED up front: + // zero trades, level untouched. Without the fix it would partial-fill. + let level = PriceLevel::new(10_000); + level + .add_order(create_sell_standard_order(1, 10_000, 1)) + .expect("maker 1"); + level + .add_order(create_reserve_order(2, 10_000, 1, 100, 1, true, Some(100))) + .expect("auto-reserve maker 2"); + level + .add_order(create_sell_standard_order(3, 10_000, u64::MAX - 2)) + .expect("maker 3 fills the visible counter to u64::MAX"); + assert_eq!(level.visible_quantity(), u64::MAX, "counter at capacity"); + + // The dry run models the abort: only maker 1's unit is reachable. + assert_eq!(level.matchable_quantity(2, Id::from_u64(999)), 1); + + let before = level.snapshot_by_insertion_seq(); + let result = level.match_order( + 2, + Id::from_u64(999), + TimeInForce::Fok, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &new_trade_id_generator(), + ); + + assert!( + result.was_killed(), + "FOK must be killed, not partially filled" + ); + assert_eq!(result.trades().len(), 0, "a killed FOK emits zero trades"); + assert_eq!(result.remaining_quantity().as_u64(), 2); + // Level byte-identical: all three makers still rest in order. + let after = level.snapshot_by_insertion_seq(); + assert_eq!( + before.iter().map(|o| o.id()).collect::>(), + after.iter().map(|o| o.id()).collect::>(), + "the queue must be untouched" + ); + assert_eq!(level.visible_quantity(), u64::MAX); + assert_counters_match_queue(&level); + } + + #[test] + fn test_level_poisoned_fails_fast() { + // Issue #130: a poisoned fill-or-kill guard (a holder panicked) fails + // fast — add_order / update_order return InvalidOperation and match_order + // refuses to match — while snapshot stays allowed for diagnostics. + let level = PriceLevel::new(10_000); + level + .add_order(create_sell_standard_order(1, 10_000, 10)) + .expect("add before poison"); + + // Genuinely poison the guard (panic while holding the write side). + level.test_poison_guard(); + + // The next admission acquires the guard, recovers the poison, sets the + // sticky flag, and fails fast. + let err = level.add_order(create_sell_standard_order(2, 10_000, 5)); + assert!( + matches!(err, Err(PriceLevelError::InvalidOperation { .. })), + "add_order must fail fast on a poisoned level, got {err:?}" + ); + + // update_order (cancel) also fails fast. + let err = level.update_order(OrderUpdate::Cancel { + order_id: Id::from_u64(1), + }); + assert!( + matches!(err, Err(PriceLevelError::InvalidOperation { .. })), + "update_order must fail fast on a poisoned level, got {err:?}" + ); + + // match_order refuses to match (empty result, no trades). + let result = level.match_order( + 5, + Id::from_u64(999), + TimeInForce::Gtc, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &new_trade_id_generator(), + ); + assert_eq!( + result.trades().len(), + 0, + "a poisoned level refuses to match" + ); + + // snapshot stays allowed (diagnostics / reconstruction): maker 1 is still + // resting and readable. + let snapshot = level.snapshot(); + assert_eq!(snapshot.order_count(), 1); + } + + #[test] + fn test_post_only_zero_trades_with_add_in_decision_window() { + // Issue #130 deterministic seam: a matchable maker is added in the EXACT + // window between PostOnly's depth decision and its commit. PostOnly must + // STILL emit zero trades (it never sweeps) and leave the added maker + // resting — no scheduler stress needed. + let level = std::sync::Arc::new(PriceLevel::new(10_000)); + let hook_level = std::sync::Arc::clone(&level); + let _hook_guard = + crate::price_level::level::set_post_only_decision_hook(Box::new(move || { + // The level was empty at the decision; add crossable depth now. + let _ = hook_level.add_order(create_sell_standard_order(1, 10_000, 50)); + })); + + let result = level.match_order( + 50, + Id::from_u64(999), + TimeInForce::Gtc, + TakerKind::PostOnly, + TimestampMs::new(1_700_000_000_000), + &new_trade_id_generator(), + ); + + assert_eq!( + result.trades().len(), + 0, + "PostOnly must emit zero trades even with depth added in the decision window" + ); + assert!( + !result.was_rejected(), + "the pre-add scan found no depth, so it rests" + ); + // The maker added in the window rests, untouched. + assert_eq!(level.order_count(), 1); + assert_counters_match_queue(&level); + } } #[cfg(test)]