From 6d4fcb27b1c6bd08906baaeaf11b3b041132d289 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 12:23:38 +0200 Subject: [PATCH 1/4] fix(price_level)!: enforce level admission and trade topology invariants add_order accepted any order without checking that its price equals the level price or that all makers share one side. A wrong-price maker traded at PriceLevel::price rather than its own stored price, and mixed maker sides produced contradictory taker sides across a single MatchResult (the taker side is derived per-maker). The maker != taker guard was a debug_assert, so release builds could emit self-trades. Admission now validates topology before any counter reservation: a mismatching price or an incompatible side is rejected with InvalidOperation and the level is left byte-identical. The level side is derived from the resting orders (first admitted maker defines it; a drained level accepts either side again) -- a correctness invariant on single-logical-writer admission, with both residual race windows documented (empty-level admission race, and the upsize remove+push transient that #119's in-place re-sequencing closes). Self-trades are now structurally impossible in every build profile: a front maker whose id equals the taker id is parked like a set-aside maker (no trade, no counter movement, makers behind it still match in FIFO). This is order-id identity -- an order can never match itself; account-level STP via user_id remains the composing book's job. BREAKING: matchable_quantity takes the taker id so the fill-or-kill dry run applies the identical skip and can never diverge from the sweep (migration guide updated, README regenerated). from_snapshot validates the same price/side topology; the infallible From<&PriceLevelSnapshot> is documented as the trusted-input path that skips validation. Closes #120 --- README.md | 35 ++++ src/lib.rs | 35 ++++ src/price_level/level.rs | 210 ++++++++++++++++++-- src/price_level/tests/level.rs | 349 ++++++++++++++++++++++++++++++--- 4 files changed, 583 insertions(+), 46 deletions(-) diff --git a/README.md b/README.md index e00794d..5f4d808 100644 --- a/README.md +++ b/README.md @@ -545,6 +545,41 @@ that could desync a level's counters from its queue: - **`OrderQueue::from_vec` is now `pub(crate)`.** It is a keep-first constructor that drops duplicates silently; the public restore path is [`PriceLevel::from_snapshot`], which rejects them. +### Migration Guide (level topology invariants — breaking) + +A [`PriceLevel`] now enforces that every resting order sits at the level's +price and shares a single side (the first admitted maker pins the side; a +fully drained level accepts either side again). [`PriceLevel::add_order`] +returns [`PriceLevelError::InvalidOperation`] for an order whose price does +not match the level, or whose side is incompatible with the resting side, +and [`PriceLevel::from_snapshot`] rejects a snapshot that violates either +(previously such orders were admitted, trading at the level price rather +than their own and producing contradictory taker sides in one +[`MatchResult`]). Callers that composed a level from mixed-price or +mixed-side orders must route each order to the correct level. + +Single-side coherence is a **correctness invariant**, not an +eventually-consistent one like the advisory counters: it holds only when a +given level's admissions arrive from a single logical writer (the composing +order book routes each price to one admission path). The side is derived +from the live queue, so under genuinely concurrent multi-writer admission a +narrow race — an opposite side slipping into a momentarily empty level — can +still admit a mixed side; see the note on the [`PriceLevel`] type. + +[`PriceLevel::matchable_quantity`] gains a `taker_id` parameter: +`matchable_quantity(incoming_quantity)` becomes +`matchable_quantity(incoming_quantity, taker_id)`. A resting maker sharing +the taker id is skipped (self-trade prevention), matching the sweep, so a +fill-or-kill dry run and the real sweep agree. `match_order` applies the +same **self-trade skip** deterministically in every build profile (it used +to be a debug-only assertion): a resting maker whose id equals the taker's +is skipped — no self-trade is emitted and the other makers still match. + +This self-trade guard is **order-id identity** — an order can never match +itself. It is NOT account/owner-level self-trade prevention: two distinct +order ids owned by the same `user_id` will still trade. Account-level STP is +the responsibility of the order book composing these levels, which owns the +account relationships a single price level does not. ## Setup Instructions diff --git a/src/lib.rs b/src/lib.rs index c59f8b6..e39dd6b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -537,6 +537,41 @@ //! - **`OrderQueue::from_vec` is now `pub(crate)`.** It is a keep-first //! constructor that drops duplicates silently; the public restore path is //! [`PriceLevel::from_snapshot`], which rejects them. +//! ## Migration Guide (level topology invariants — breaking) +//! +//! A [`PriceLevel`] now enforces that every resting order sits at the level's +//! price and shares a single side (the first admitted maker pins the side; a +//! fully drained level accepts either side again). [`PriceLevel::add_order`] +//! returns [`PriceLevelError::InvalidOperation`] for an order whose price does +//! not match the level, or whose side is incompatible with the resting side, +//! and [`PriceLevel::from_snapshot`] rejects a snapshot that violates either +//! (previously such orders were admitted, trading at the level price rather +//! than their own and producing contradictory taker sides in one +//! [`MatchResult`]). Callers that composed a level from mixed-price or +//! mixed-side orders must route each order to the correct level. +//! +//! Single-side coherence is a **correctness invariant**, not an +//! eventually-consistent one like the advisory counters: it holds only when a +//! given level's admissions arrive from a single logical writer (the composing +//! order book routes each price to one admission path). The side is derived +//! from the live queue, so under genuinely concurrent multi-writer admission a +//! narrow race — an opposite side slipping into a momentarily empty level — can +//! still admit a mixed side; see the note on the [`PriceLevel`] type. +//! +//! [`PriceLevel::matchable_quantity`] gains a `taker_id` parameter: +//! `matchable_quantity(incoming_quantity)` becomes +//! `matchable_quantity(incoming_quantity, taker_id)`. A resting maker sharing +//! the taker id is skipped (self-trade prevention), matching the sweep, so a +//! fill-or-kill dry run and the real sweep agree. `match_order` applies the +//! same **self-trade skip** deterministically in every build profile (it used +//! to be a debug-only assertion): a resting maker whose id equals the taker's +//! is skipped — no self-trade is emitted and the other makers still match. +//! +//! This self-trade guard is **order-id identity** — an order can never match +//! itself. It is NOT account/owner-level self-trade prevention: two distinct +//! order ids owned by the same `user_id` will still trade. Account-level STP is +//! the responsibility of the order book composing these levels, which owns the +//! account relationships a single price level does not. //! mod orders; diff --git a/src/price_level/level.rs b/src/price_level/level.rs index cc1116c..4145f2a 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -14,7 +14,33 @@ use std::str::FromStr; use std::sync::Arc; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; -/// A lock-free implementation of a price level in a limit order book +/// A lock-free implementation of a price level in a limit order book. +/// +/// # Topology +/// +/// Every resting order sits at [`Self::price`] and shares a single side. The +/// side is not stored: it is derived from the resting orders themselves (the +/// first admitted maker defines it, and a drained level accepts either side +/// again), so [`Self::add_order`] rejects a mismatching price or side. Deriving +/// the side is lock-free. +/// +/// Single-side coherence is a **correctness invariant**, not an +/// eventually-consistent one — unlike the advisory quantity / count counters +/// (issue #68), a level that admits two makers of opposite sides does not +/// converge to a correct state later. Upholding it therefore relies on a +/// strictly stronger assumption than the counters: that a given level's +/// admissions arrive from a **single logical writer** (the composing order +/// book routes each price to one admission path). Absent that, two races can +/// admit an incompatible side: +/// +/// - Two **opposite-side admissions into a genuinely empty level**: both may +/// observe no resting order and both admit. +/// - An **opposite-side admission racing a same-side upsize** on a +/// single-order level: a quantity increase via [`Self::update_order`] +/// re-sequences the sole maker with a remove-then-push, transiently emptying +/// the queue, and an opposite-side admission observing that gap admits. This +/// window rides on the same remove+push gap that issue #119's in-place +/// re-sequencing closes. #[derive(Debug)] pub struct PriceLevel { /// The price of this level @@ -71,6 +97,38 @@ impl PriceLevel { } } + // Topology invariants (same rules `add_order` enforces): every order + // must sit at the level's price and share a single side. A snapshot that + // violates either would reconstruct a level that trades at the wrong + // price or emits contradictory taker sides, so reject it here rather + // than restore an incoherent level. + { + let level_price = snapshot.price().as_u128(); + let mut level_side = None; + for order in snapshot.orders() { + if order.price().as_u128() != level_price { + return Err(PriceLevelError::InvalidOperation { + message: format!( + "snapshot order price {} does not match level price {level_price}", + order.price().as_u128() + ), + }); + } + match level_side { + None => level_side = Some(order.side()), + Some(side) if side != order.side() => { + return Err(PriceLevelError::InvalidOperation { + message: format!( + "snapshot order side {:?} is incompatible with the level side {side:?}", + order.side() + ), + }); + } + Some(_) => {} + } + } + } + let order_count = snapshot.orders().len(); let visible_quantity = snapshot.visible_quantity().as_u64(); let hidden_quantity = snapshot.hidden_quantity().as_u64(); @@ -254,15 +312,58 @@ impl PriceLevel { /// undo on these advisory counters), so `try_push_with` publishes nothing /// and no counter is left drifted. /// + /// # Topology invariants + /// + /// A level holds orders at exactly one price and one side. The order's price + /// must equal the level's price, and its side must match the side of the + /// orders already resting here (the first admitted maker pins the side; a + /// fully drained level accepts either side again). Both are checked before + /// any counter is touched, so a rejected order leaves the level unchanged. + /// /// # Errors /// - /// Returns [`PriceLevelError::InvalidOperation`] if the order's own - /// visible + hidden total overflows `u64`, or if admitting it would - /// overflow the level's visible-quantity, hidden-quantity, or order-count - /// counter, or [`PriceLevelError::DuplicateOrderId`] if an order with the - /// same id already rests at this level. A duplicate id takes precedence over - /// a counter overflow. In every case the level is unchanged. + /// Returns [`PriceLevelError::InvalidOperation`] if the order's price does + /// not match the level's, if its side is incompatible with the resting + /// side, if the order's own visible + hidden total overflows `u64`, or if + /// admitting it would overflow the level's visible-quantity, + /// hidden-quantity, or order-count counter; or + /// [`PriceLevelError::DuplicateOrderId`] if an order with the same id + /// 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> { + // -------- Admission topology invariants (cheapest checks, no mutation) -------- + // + // A level holds orders at exactly one price and one side. Reject a + // mismatch BEFORE reserving any counter capacity, so the level is left + // completely unchanged. Price is the cheapest check (two `u128`s), so it + // goes first; the side is derived from whatever is already resting. + if order.price().as_u128() != self.price { + return Err(PriceLevelError::InvalidOperation { + message: format!( + "order price {} does not match level price {}", + order.price().as_u128(), + self.price + ), + }); + } + // The level's side is defined by its resting makers: the first maker + // pins it, and a drained (empty) level accepts either side again — the + // queue is the source of truth, so no separate side state has to be + // stored or reset. Any resting order's side is representative (this very + // invariant keeps them all equal). Deriving it is lock-free; see the + // type-level note on the one residual window (two opposite-side + // admissions racing into a genuinely empty level). + if let Some(resting_side) = self.orders.iter_orders().next().map(|o| o.side()) + && order.side() != resting_side + { + return Err(PriceLevelError::InvalidOperation { + message: format!( + "order side {:?} is incompatible with the level's resting side {resting_side:?}", + order.side() + ), + }); + } + // Calculate quantities. let visible_qty = order.visible_quantity().as_u64(); let hidden_qty = order.hidden_quantity().as_u64(); @@ -423,8 +524,13 @@ impl PriceLevel { /// level. In particular a zero-visible iceberg (or auto-replenishing /// reserve) backed by hidden quantity counts as matchable depth, because the /// sweep will draw that hidden into visible and fill it. - fn has_matchable_depth(&self) -> bool { - self.iter_orders().any(|order| order.is_matchable()) + /// + /// A resting maker sharing `taker_id` is ignored: the sweep skips it for + /// 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()) } /// Computes how much of `incoming_quantity` this level could actually fill @@ -439,6 +545,11 @@ impl PriceLevel { /// consume — never an over- or under-count — which is what fill-or-kill /// (all-or-nothing) correctly depends on. /// + /// `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 + /// skips it for self-trade prevention, so the prediction and the sweep can + /// never diverge. + /// /// It allocates a working snapshot and is only used on the cold /// fill-or-kill path, not on the hot `Gtc` sweep. /// @@ -447,7 +558,7 @@ impl PriceLevel { /// feasibility instead of re-deriving the sweep, which would risk drifting /// from the real `match_order` behavior. #[must_use] - pub fn matchable_quantity(&self, incoming_quantity: u64) -> u64 { + pub fn matchable_quantity(&self, incoming_quantity: u64, taker_id: Id) -> u64 { if incoming_quantity == 0 { return 0; } @@ -469,6 +580,12 @@ impl PriceLevel { let Some(order) = pending.pop_front() else { break; }; + // Self-trade prevention parity: the real sweep skips a maker sharing + // the taker id (`SelfTradeSkipped`), so the dry run must skip it too, + // or fill-or-kill would predict depth the sweep will not take. + if order.id() == taker_id { + continue; + } let (consumed, updated_order, hidden_reduced, new_remaining) = order.match_against(remaining); @@ -611,7 +728,12 @@ impl PriceLevel { /// 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. + /// `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. pub fn match_order( &self, incoming_quantity: u64, @@ -627,7 +749,10 @@ impl PriceLevel { // 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() { + 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, @@ -642,7 +767,7 @@ impl PriceLevel { // 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 { - let available = self.matchable_quantity(incoming_quantity); + let available = self.matchable_quantity(incoming_quantity, taker_order_id); if available < incoming_quantity { tracing::debug!( taker_order_id = %taker_order_id, @@ -715,19 +840,24 @@ impl PriceLevel { new_remaining: u64, } - // Either the maker progressed (carrying `StepData`), was parked by the - // no-progress guard (`SetAside`), or forced the sweep to abort because + // Either the maker progressed (carrying `StepData`), was parked + // (`SetAside` for the no-progress guard, `SelfTradeSkipped` for a maker + // whose id equals the taker's), or forced the sweep to abort because // committing its replenishment would overflow the level's visible // counter (`Abort`). The parked / abort variants thread the maker's id - // (and, for `SetAside`, its insertion seq) OUT of the locked decision - // closure so the `warn!` / `error!` can name the maker without logging - // inside the per-entry lock. + // (and, where relevant, its insertion seq) OUT of the locked decision + // closure so the caller's `warn!` / `debug!` can name the maker without + // logging inside the per-entry lock. enum StepResult { Progressed(StepData), SetAside { maker_id: Id, seq: u64, }, + SelfTradeSkipped { + maker_id: Id, + seq: u64, + }, /// The FIFO-front maker would replenish, but moving the drawn hidden /// tranche into the level's visible counter would take it past /// `u64::MAX` — a depth the level cannot represent. The maker is left @@ -741,6 +871,29 @@ impl PriceLevel { while remaining > 0 { let outcome = self.orders.match_front(&mut set_aside, |seq, order_arc| { + // Self-trade prevention (deterministic in every build profile, + // not a debug-only assert): a resting maker must never trade + // against a taker carrying the same id. Skip it — park its + // sequence like a no-progress maker so the sweep advances to the + // makers behind it — rather than emit a self-trade. The maker is + // left untouched (no trade, counters and queue unchanged). + // + // Scope: this is ORDER-ID identity — an order can never match + // *itself*. It is NOT account/owner-level self-trade prevention: + // two distinct order ids owned by the same `user_id` will still + // trade here. Account-level STP is the composing order book's + // responsibility (it knows the owner relationships this level + // does not). + if order_arc.id() == taker_order_id { + return ( + FrontAction::SetAside, + StepResult::SelfTradeSkipped { + maker_id: order_arc.id(), + seq, + }, + ); + } + let (consumed, updated_order, hidden_reduced, new_remaining) = order_arc.match_against(remaining); @@ -881,6 +1034,20 @@ impl PriceLevel { ); break; } + StepResult::SelfTradeSkipped { maker_id, seq } => { + // Self-trade prevention: the front maker shares the + // taker's id. Skip it (parked like a set-aside maker) + // and advance to the makers behind it; no trade is + // emitted and the maker is left untouched. + tracing::debug!( + price = self.price, + remaining, + order_id = %maker_id, + seq, + "match sweep: front maker shares the taker id; skipped to prevent self-trade" + ); + continue; + } StepResult::Progressed(data) => data, }; let new_remaining = data.new_remaining; @@ -897,9 +1064,10 @@ impl PriceLevel { let trade_id = Id::from_uuid(trade_id_generator.next()); - // A resting maker must never be the taker matching - // against it. Debug-only invariant of the caller. - debug_assert!(data.maker_id != taker_order_id, "self-fill: maker == taker"); + // A resting maker can never be the taker here: a maker + // sharing the taker id is skipped (`SelfTradeSkipped`) + // before it reaches this point, so no self-trade is ever + // emitted — deterministically, in every build profile. let trade = Trade::with_timestamp( trade_id, diff --git a/src/price_level/tests/level.rs b/src/price_level/tests/level.rs index 08902bb..74d26b7 100644 --- a/src/price_level/tests/level.rs +++ b/src/price_level/tests/level.rs @@ -40,7 +40,7 @@ mod tests { .add_order(create_standard_order(1, 10000, 100)) .expect("add_order should succeed"); price_level - .add_order(create_iceberg_order(2, 10000, 50, 200)) + .add_order(create_buy_iceberg_order(2, 10000, 50, 200)) .expect("add_order should succeed"); let package = price_level @@ -243,13 +243,21 @@ mod tests { .add_order(create_standard_order(1, 15000, 100)) .expect("add_order should succeed"); price_level - .add_order(create_iceberg_order(2, 15000, 40, 120)) + .add_order(create_buy_iceberg_order(2, 15000, 40, 120)) .expect("add_order should succeed"); price_level .add_order(create_post_only_order(3, 15000, 60)) .expect("add_order should succeed"); price_level - .add_order(create_reserve_order(4, 15000, 30, 90, 15, true, Some(20))) + .add_order(create_buy_reserve_order( + 4, + 15000, + 30, + 90, + 15, + true, + Some(20), + )) .expect("add_order should succeed"); let snapshot = price_level.snapshot(); @@ -284,7 +292,7 @@ mod tests { .add_order(create_standard_order(10, 17500, 80)) .expect("add_order should succeed"); price_level - .add_order(create_trailing_stop_order(11, 17500, 50)) + .add_order(create_buy_trailing_stop_order(11, 17500, 50)) .expect("add_order should succeed"); price_level .add_order(create_pegged_order(12, 17500, 40)) @@ -421,6 +429,68 @@ mod tests { } } + // Buy-side variants of the Sell-defaulting helpers, for tests that mix + // several order types at one level (issue #120: a level holds a single + // side, so every maker in these tests must share it). + fn create_buy_iceberg_order(id: u64, price: u128, visible: u64, hidden: u64) -> OrderType<()> { + let timestamp = TIMESTAMP_COUNTER.fetch_add(1, Ordering::SeqCst); + OrderType::IcebergOrder { + id: Id::from_u64(id), + price: Price::new(price), + visible_quantity: Quantity::new(visible), + hidden_quantity: Quantity::new(hidden), + side: Side::Buy, + user_id: Hash32::zero(), + timestamp: TimestampMs::new(timestamp), + time_in_force: TimeInForce::Gtc, + extra_fields: (), + } + } + + fn create_buy_trailing_stop_order(id: u64, price: u128, quantity: u64) -> OrderType<()> { + let timestamp = TIMESTAMP_COUNTER.fetch_add(1, Ordering::SeqCst); + OrderType::TrailingStop { + id: Id::from_u64(id), + price: Price::new(price), + quantity: Quantity::new(quantity), + side: Side::Buy, + user_id: Hash32::zero(), + timestamp: TimestampMs::new(timestamp), + time_in_force: TimeInForce::Gtc, + trail_amount: Quantity::new(100), + last_reference_price: Price::new(price + 100u128), + extra_fields: (), + } + } + + #[allow(clippy::too_many_arguments)] + fn create_buy_reserve_order( + id: u64, + price: u128, + visible: u64, + hidden: u64, + threshold: u64, + auto_replenish: bool, + replenish_amount: Option, + ) -> OrderType<()> { + let timestamp = TIMESTAMP_COUNTER.fetch_add(1, Ordering::SeqCst); + OrderType::ReserveOrder { + id: Id::from_u64(id), + price: Price::new(price), + visible_quantity: Quantity::new(visible), + hidden_quantity: Quantity::new(hidden), + side: Side::Buy, + user_id: Hash32::zero(), + timestamp: TimestampMs::new(timestamp), + time_in_force: TimeInForce::Gtc, + replenish_threshold: Quantity::new(threshold), + replenish_amount: replenish_amount + .map(|amount| NonZeroU64::new(amount).expect("test replenish amount must be > 0")), + auto_replenish, + extra_fields: (), + } + } + fn create_fill_or_kill_order(id: u64, price: u128, quantity: u64) -> OrderType<()> { let timestamp = TIMESTAMP_COUNTER.fetch_add(1, Ordering::SeqCst); OrderType::Standard { @@ -532,13 +602,13 @@ mod tests { .add_order(create_standard_order(1, 10000, 100)) .expect("add_order should succeed"); price_level - .add_order(create_iceberg_order(2, 10000, 50, 200)) + .add_order(create_buy_iceberg_order(2, 10000, 50, 200)) .expect("add_order should succeed"); price_level .add_order(create_post_only_order(3, 10000, 75)) .expect("add_order should succeed"); price_level - .add_order(create_reserve_order(4, 10000, 25, 100, 100, true, None)) + .add_order(create_buy_reserve_order(4, 10000, 25, 100, 100, true, None)) .expect("add_order should succeed"); assert_eq!(price_level.visible_quantity(), 250); // 100 + 50 + 75 + 25 @@ -558,7 +628,7 @@ mod tests { .add_order(create_standard_order(1, 10000, 100)) .expect("add_order should succeed"); price_level - .add_order(create_iceberg_order(2, 10000, 50, 200)) + .add_order(create_buy_iceberg_order(2, 10000, 50, 200)) .expect("add_order should succeed"); // Cancel the standard order using OrderUpdate @@ -607,7 +677,7 @@ mod tests { .add_order(create_standard_order(1, 10000, 100)) .expect("add_order should succeed"); price_level - .add_order(create_iceberg_order(2, 10000, 50, 200)) + .add_order(create_buy_iceberg_order(2, 10000, 50, 200)) .expect("add_order should succeed"); let orders = price_level.snapshot_orders(); @@ -2268,15 +2338,17 @@ mod tests { #[test] fn test_update_quantity_trailing_stop_decrease_keeps_position() { + // Buy side to match the plain maker the helper queues behind it (a level + // holds a single side, issue #120). assert_update_quantity_decrease_keeps_position( - create_trailing_stop_order(1, 10000, 100), + create_buy_trailing_stop_order(1, 10000, 100), 40, ); } #[test] fn test_update_quantity_trailing_stop_increase_demotes() { - assert_update_quantity_increase_demotes(create_trailing_stop_order(1, 10000, 100), 150); + assert_update_quantity_increase_demotes(create_buy_trailing_stop_order(1, 10000, 100), 150); } #[test] @@ -3208,7 +3280,7 @@ mod tests { .add_order(create_standard_order(2, 10000, 100)) .expect("add_order should succeed"); price_level - .add_order(create_iceberg_order(3, 10000, 50, 200)) + .add_order(create_buy_iceberg_order(3, 10000, 50, 200)) .expect("add_order should succeed"); // Decrease (in place) on a standard order. @@ -3459,13 +3531,13 @@ mod tests { .add_order(create_good_till_date_order(3, 10000, 100, 1617000000000)) .expect("add_order should succeed"); price_level - .add_order(create_reserve_order(4, 10000, 100, 100, 20, true, None)) + .add_order(create_buy_reserve_order(4, 10000, 100, 100, 20, true, None)) .expect("add_order should succeed"); price_level - .add_order(create_iceberg_order(5, 10000, 50, 100)) + .add_order(create_buy_iceberg_order(5, 10000, 50, 100)) .expect("add_order should succeed"); - let input = "PriceLevel:price=10000;visible_quantity=375;hidden_quantity=200;order_count=5;orders=[Standard:id=00000000-0000-0001-0000-000000000000;price=10000;quantity=50;side=BUY;timestamp=1616823000000;time_in_force=GTC,Standard:id=00000000-0000-0002-0000-000000000000;price=10000;quantity=75;side=BUY;timestamp=1616823000001;time_in_force=GTC,Standard:id=00000000-0000-0003-0000-000000000000;price=10000;quantity=100;side=BUY;timestamp=1616823000002;time_in_force=GTD-1617000000000,ReserveOrder:id=00000000-0000-0004-0000-000000000000;price=10000;visible_quantity=100;hidden_quantity=100;side=SELL;timestamp=1616823000003;time_in_force=GTC;replenish_threshold=20;replenish_amount=None;auto_replenish=true,IcebergOrder:id=00000000-0000-0005-0000-000000000000;price=10000;visible_quantity=50;hidden_quantity=100;side=SELL;timestamp=1616823000004;time_in_force=GTC]"; + let input = "PriceLevel:price=10000;visible_quantity=375;hidden_quantity=200;order_count=5;orders=[Standard:id=00000000-0000-0001-0000-000000000000;price=10000;quantity=50;side=BUY;timestamp=1616823000000;time_in_force=GTC,Standard:id=00000000-0000-0002-0000-000000000000;price=10000;quantity=75;side=BUY;timestamp=1616823000001;time_in_force=GTC,Standard:id=00000000-0000-0003-0000-000000000000;price=10000;quantity=100;side=BUY;timestamp=1616823000002;time_in_force=GTD-1617000000000,ReserveOrder:id=00000000-0000-0004-0000-000000000000;price=10000;visible_quantity=100;hidden_quantity=100;side=BUY;timestamp=1616823000003;time_in_force=GTC;replenish_threshold=20;replenish_amount=None;auto_replenish=true,IcebergOrder:id=00000000-0000-0005-0000-000000000000;price=10000;visible_quantity=50;hidden_quantity=100;side=BUY;timestamp=1616823000004;time_in_force=GTC]"; let result = PriceLevel::from_str(input); if let Err(ref err) = result { @@ -3661,7 +3733,7 @@ mod tests { .add_order(create_standard_order(1, 10000, 100)) .expect("add_order should succeed"); price_level - .add_order(create_iceberg_order(2, 10000, 50, 150)) + .add_order(create_buy_iceberg_order(2, 10000, 50, 150)) .expect("add_order should succeed"); // Serialize to JSON @@ -4259,7 +4331,7 @@ mod tests { .expect("add_order should succeed"); } else { level - .add_order(create_iceberg_order( + .add_order(create_buy_iceberg_order( id, PRICE, 1 + (base % 5), @@ -5994,7 +6066,7 @@ mod tests { .add_order(create_standard_order(2, 10_000, 50)) .expect("add_order should succeed"); level - .add_order(create_iceberg_order(3, 10_000, 20, 30)) + .add_order(create_buy_iceberg_order(3, 10_000, 20, 30)) .expect("add_order should succeed"); let owned: Vec = level @@ -6083,10 +6155,19 @@ mod tests { .add_order(create_standard_order(2, 10_000, 50)) .expect("add_order should succeed"); - assert_eq!(level.matchable_quantity(0), 0, "zero taker fills nothing"); - assert_eq!(level.matchable_quantity(120), 120, "taker below depth"); + let taker = Id::from_u64(999); + assert_eq!( + level.matchable_quantity(0, taker), + 0, + "zero taker fills nothing" + ); + assert_eq!( + level.matchable_quantity(120, taker), + 120, + "taker below depth" + ); // A taker above the available depth is capped at the depth. - let predicted = level.matchable_quantity(200); + let predicted = level.matchable_quantity(200, taker); assert_eq!(predicted, 150, "taker above depth is capped at depth"); // The dry run does not mutate, so the real sweep on the same level must @@ -6111,7 +6192,7 @@ mod tests { let ice = PriceLevel::new(10_000); ice.add_order(create_iceberg_order(1, 10_000, 10, 40)) .expect("add_order should succeed"); - let predicted_ice = ice.matchable_quantity(100); + let predicted_ice = ice.matchable_quantity(100, Id::from_u64(998)); assert_eq!( predicted_ice, 50, "matchable_quantity reaches hidden depth via replenishment" @@ -6433,7 +6514,9 @@ mod tests { )) .expect("reserve own total fits u64"); level - .add_order(create_standard_order(2, 10_000, u64::MAX - 1)) + // Sell to stay side-coherent with the reserve above (issue #120 + // pins the level side to its first resting maker). + .add_order(create_sell_standard_order(2, 10_000, u64::MAX - 1)) .expect("standard own total fits u64"); assert_eq!(level.visible_quantity(), u64::MAX); @@ -6643,8 +6726,8 @@ mod tests { // The same id as a DIFFERENT order variant is still a duplicate. let duplicates = [ - create_iceberg_order(1, 10_000, 50, 50), - create_reserve_order(1, 10_000, 30, 60, 10, true, Some(20)), + create_buy_iceberg_order(1, 10_000, 50, 50), + create_buy_reserve_order(1, 10_000, 30, 60, 10, true, Some(20)), create_standard_order(1, 10_000, 5), ]; for dup in duplicates { @@ -6665,7 +6748,7 @@ mod tests { // A genuinely distinct id still admits fine. level - .add_order(create_iceberg_order(2, 10_000, 50, 50)) + .add_order(create_buy_iceberg_order(2, 10_000, 50, 50)) .expect("distinct id must admit"); assert_eq!(level.order_count(), 2); } @@ -6836,6 +6919,33 @@ mod tests { ); } + // ------------------------------------------------------------------ + // Issue #120 — admission and trade topology invariants + // ------------------------------------------------------------------ + + #[test] + fn test_add_order_wrong_price_rejected() { + let level = PriceLevel::new(10_000); + level + .add_order(create_standard_order(1, 10_000, 100)) + .expect("in-price admission must succeed"); + + let before = level.snapshot_to_json().expect("snapshot before"); + // An order at a different price must be rejected, level unchanged. + match level.add_order(create_standard_order(2, 10_001, 50)) { + Err(PriceLevelError::InvalidOperation { message }) => { + assert!(message.contains("price"), "unexpected message: {message}"); + } + other => panic!("expected wrong-price InvalidOperation, got {other:?}"), + } + assert_eq!(level.order_count(), 1); + assert_eq!( + level.snapshot_to_json().expect("snapshot after"), + before, + "a rejected wrong-price admission must leave the level unchanged" + ); + } + #[test] fn test_try_from_snapshot_propagates_duplicate_order_id() { // Finding 3 (PR #125): the infallible `From<&PriceLevelSnapshot>` (which @@ -6870,6 +6980,195 @@ mod tests { assert_eq!(restored.order_count(), 2); assert_eq!(restored.visible_quantity(), 30); } + + #[test] + fn test_add_order_mixed_side_rejected_then_readmissible_after_drain() { + let level = PriceLevel::new(10_000); + // First maker pins the level side to Buy. + level + .add_order(create_standard_order(1, 10_000, 100)) + .expect("first (Buy) admission must succeed"); + + let before = level.snapshot_to_json().expect("snapshot before"); + // A Sell maker is incompatible with the Buy level. + match level.add_order(create_sell_standard_order(2, 10_000, 50)) { + Err(PriceLevelError::InvalidOperation { message }) => { + assert!(message.contains("side"), "unexpected message: {message}"); + } + other => panic!("expected mixed-side InvalidOperation, got {other:?}"), + } + assert_eq!(level.order_count(), 1); + assert_eq!( + level.snapshot_to_json().expect("snapshot after"), + before, + "a rejected mixed-side admission must leave the level unchanged" + ); + + // Drain the level to empty via a full match. + let namespace = Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap(); + let generator = UuidGenerator::new(namespace); + let _ = level.match_order( + 100, + Id::from_u64(900), + TimeInForce::Gtc, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ); + assert_eq!(level.order_count(), 0, "the level must be drained empty"); + + // A drained level accepts either side again: the opposite side now admits. + level + .add_order(create_sell_standard_order(3, 10_000, 70)) + .expect("a drained level must re-accept the opposite side"); + assert_eq!(level.order_count(), 1); + } + + #[test] + fn test_match_order_self_trade_skipped() { + // Makers 1, 2, 3 rest in FIFO order; the taker shares maker 1's id. + let level = PriceLevel::new(10_000); + level + .add_order(create_standard_order(1, 10_000, 40)) + .expect("maker 1 admits"); + level + .add_order(create_standard_order(2, 10_000, 30)) + .expect("maker 2 admits"); + level + .add_order(create_standard_order(3, 10_000, 50)) + .expect("maker 3 admits"); + + let namespace = Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap(); + let generator = UuidGenerator::new(namespace); + // Taker id == maker 1's id: maker 1 must be skipped, 2 and 3 consumed. + let result = level.match_order( + 1_000, + Id::from_u64(1), + TimeInForce::Gtc, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ); + + // No trade names the taker as its own maker (no self-trade emitted). + let makers: Vec = result + .trades() + .as_vec() + .iter() + .map(|t| t.maker_order_id()) + .collect(); + assert!( + makers.iter().all(|m| *m != Id::from_u64(1)), + "no trade may have maker == taker; got {makers:?}" + ); + // The other makers are still consumed, in FIFO order. + assert_eq!(makers, vec![Id::from_u64(2), Id::from_u64(3)]); + // MatchResult stays consistent: executed == 30 + 50 = 80. + assert_eq!( + result.executed_quantity().expect("no overflow").as_u64(), + 80 + ); + // The skipped maker 1 is left resting, untouched. + let resting: Vec = level + .snapshot_by_insertion_seq() + .iter() + .map(|o| o.id()) + .collect(); + assert_eq!(resting, vec![Id::from_u64(1)]); + } + + #[test] + fn test_matchable_quantity_self_trade_parity_with_fok() { + // Maker 1 (shares the taker id) has 40; maker 2 has 60. A taker id 1 + // can only take maker 2's 60 (maker 1 is self-trade-skipped). + let level = PriceLevel::new(10_000); + level + .add_order(create_standard_order(1, 10_000, 40)) + .expect("maker 1 admits"); + level + .add_order(create_standard_order(2, 10_000, 60)) + .expect("maker 2 admits"); + + // The dry run agrees: skipping the self-trade maker leaves 60. + assert_eq!(level.matchable_quantity(100, Id::from_u64(1)), 60); + assert_eq!(level.matchable_quantity(60, Id::from_u64(1)), 60); + + let namespace = Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap(); + // A FOK taker (id 1) of 100 must be KILLED: only 60 is takeable. + let killed = level.match_order( + 100, + Id::from_u64(1), + TimeInForce::Fok, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &UuidGenerator::new(namespace), + ); + assert!( + killed.was_killed(), + "FOK 100 must be killed (only 60 takeable)" + ); + assert_eq!(killed.trades().len(), 0); + assert_eq!( + level.order_count(), + 2, + "a killed FOK leaves the queue untouched" + ); + + // A FOK taker (id 1) of 60 must FILL: exactly maker 2's 60. + let filled = level.match_order( + 60, + Id::from_u64(1), + TimeInForce::Fok, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_001), + &UuidGenerator::new(namespace), + ); + assert!(filled.is_complete(), "FOK 60 must fill"); + let makers: Vec = filled + .trades() + .as_vec() + .iter() + .map(|t| t.maker_order_id()) + .collect(); + assert_eq!(makers, vec![Id::from_u64(2)], "only maker 2 fills the FOK"); + } + + #[test] + fn test_from_snapshot_rejects_wrong_price_and_mixed_side() { + // Wrong price: an order whose price differs from the level's. + let wrong_price = crate::price_level::PriceLevelSnapshot::with_orders( + Price::new(10_000), + vec![ + std::sync::Arc::new(create_standard_order(1, 10_000, 100)), + std::sync::Arc::new(create_standard_order(2, 10_001, 50)), + ], + ) + .expect("snapshot construction succeeds"); + assert!( + matches!( + PriceLevel::from_snapshot(wrong_price), + Err(PriceLevelError::InvalidOperation { .. }) + ), + "from_snapshot must reject a wrong-price order" + ); + + // Mixed side: Buy and Sell orders in one snapshot. + let mixed_side = crate::price_level::PriceLevelSnapshot::with_orders( + Price::new(10_000), + vec![ + std::sync::Arc::new(create_standard_order(1, 10_000, 100)), + std::sync::Arc::new(create_sell_standard_order(2, 10_000, 50)), + ], + ) + .expect("snapshot construction succeeds"); + assert!( + matches!( + PriceLevel::from_snapshot(mixed_side), + Err(PriceLevelError::InvalidOperation { .. }) + ), + "from_snapshot must reject a mixed-side snapshot" + ); + } } #[cfg(test)] From e25b4e4afea30072968aed7c138730d01dc0fa32 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 17:02:55 +0200 Subject: [PATCH 2/4] address review: atomic side pin, snapshot epoch, terminal self-match - side and order count now live in one packed atomic word: admission establishes the side / joins / rejects in a single CAS, and removal decrements and un-pins in the same CAS when the count reaches zero, so the two races derive-from-queue left open (opposite-side admissions into an empty level; an opposite-side admission racing an upsize's transient queue gap) are closed by construction -- and every admission drops the per-call queue iteration, cutting add_order latency ~39% in the admission bench - a topology epoch bumps on every pin / un-pin and snapshot() retries its materialization if the epoch moves, so a checksummed snapshot can never capture a torn old-side/new-side view across a valid drain-then-re-admit transition (single-side backstop bounds the retry) - a taker whose id already rests at the level is now rejected terminally before any sweep for every TIF -- zero trades, level byte-identical -- per the issue's no-mutation self-match contract; the in-sweep skip remains as defense-in-depth for admissions racing the sweep - the set-aside bookkeeping insert moved out from under the shard lock (no allocation under the entry lock; no eager pre-reserve either -- the common zero-set-aside match stays allocation-free) --- src/price_level/level.rs | 449 +++++++++++++++++++++++++++------ src/price_level/order_queue.rs | 43 +++- src/price_level/tests/level.rs | 326 ++++++++++++++++++------ 3 files changed, 652 insertions(+), 166 deletions(-) diff --git a/src/price_level/level.rs b/src/price_level/level.rs index 4145f2a..3372bbe 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -3,7 +3,7 @@ use crate::UuidGenerator; use crate::errors::PriceLevelError; use crate::execution::{MatchResult, TakerKind, Trade}; -use crate::orders::{Id, OrderType, OrderUpdate, TimeInForce}; +use crate::orders::{Id, OrderType, OrderUpdate, Side, TimeInForce}; use crate::price_level::order_queue::{FrontAction, FrontOutcome, OrderQueue}; use crate::price_level::{PriceLevelSnapshot, PriceLevelSnapshotPackage, PriceLevelStatistics}; use crate::utils::{Price, Quantity, TimestampMs}; @@ -12,35 +12,91 @@ use std::fmt::Display; use std::str::FromStr; use std::sync::Arc; -use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU64, Ordering}; + +/// 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 +/// both into one atomic makes the side pin and the count move together in a +/// single compare-exchange, so a drain's un-pin can never race an admission's +/// pin across two independent atomics. +mod topology { + use crate::orders::Side; + + /// Bits reserved for the resting-order count (the rest hold the side tag). + /// `u64::MAX >> 2` orders is astronomically beyond any level's capacity, so + /// nothing is lost by borrowing the top two bits for the tag. + pub(super) const COUNT_BITS: u32 = 62; + pub(super) const COUNT_MASK: u64 = (1 << COUNT_BITS) - 1; + pub(super) const TAG_UNPINNED: u64 = 0; + pub(super) const TAG_BUY: u64 = 1; + pub(super) const TAG_SELL: u64 = 2; + + #[inline] + pub(super) fn tag_of(side: Side) -> u64 { + match side { + Side::Buy => TAG_BUY, + Side::Sell => TAG_SELL, + } + } + + #[inline] + pub(super) fn side_of_tag(tag: u64) -> Option { + match tag { + TAG_BUY => Some(Side::Buy), + TAG_SELL => Some(Side::Sell), + _ => None, + } + } + + #[inline] + pub(super) fn pack(tag: u64, count: u64) -> u64 { + (tag << COUNT_BITS) | count + } + + #[inline] + pub(super) fn tag(word: u64) -> u64 { + word >> COUNT_BITS + } + + #[inline] + pub(super) fn count(word: u64) -> u64 { + word & COUNT_MASK + } +} /// A lock-free implementation of a price level in a limit order book. /// /// # Topology /// /// Every resting order sits at [`Self::price`] and shares a single side. The -/// side is not stored: it is derived from the resting orders themselves (the -/// first admitted maker defines it, and a drained level accepts either side -/// again), so [`Self::add_order`] rejects a mismatching price or side. Deriving -/// the side is lock-free. +/// side is **pinned atomically** rather than derived from the queue: a single +/// `topology` word packs `(pinned side, resting order count)` so that an +/// admission's side decision and the drain that un-pins an emptied level are +/// one compare-exchange, never two racing atomics (issue #126). The first +/// admitted maker pins the side; each later same-side admission bumps the +/// count under the same CAS; the removal that brings the count to zero un-pins +/// in the same CAS, so a fully drained level accepts either side again. An +/// opposite-side admission into a non-empty level is rejected. /// /// Single-side coherence is a **correctness invariant**, not an /// eventually-consistent one — unlike the advisory quantity / count counters -/// (issue #68), a level that admits two makers of opposite sides does not -/// converge to a correct state later. Upholding it therefore relies on a -/// strictly stronger assumption than the counters: that a given level's -/// admissions arrive from a **single logical writer** (the composing order -/// book routes each price to one admission path). Absent that, two races can -/// admit an incompatible side: +/// (issue #68), a level that admitted two makers of opposite sides would not +/// converge to a correct state later. Pinning side+count in one atomic upholds +/// it under **arbitrary concurrent admissions and removals**, closing both +/// races the earlier derive-from-queue scheme left open: +/// +/// - Two **opposite-side admissions into a genuinely empty level** now +/// serialize on the pin CAS: exactly one establishes the side, the other +/// observes a non-empty opposite-side level and is rejected. +/// - An **opposite-side admission racing a same-side upsize** cannot slip +/// through a transient queue gap: the pin persists across a maker's +/// remove-then-re-add because the count never reaches zero, so the side stays +/// pinned even while the queue momentarily looks empty. /// -/// - Two **opposite-side admissions into a genuinely empty level**: both may -/// observe no resting order and both admit. -/// - An **opposite-side admission racing a same-side upsize** on a -/// single-order level: a quantity increase via [`Self::update_order`] -/// re-sequences the sole maker with a remove-then-push, transiently emptying -/// the queue, and an opposite-side admission observing that gap admits. This -/// window rides on the same remove+push gap that issue #119's in-place -/// re-sequencing closes. +/// A concurrent [`Self::snapshot`] cannot capture a torn old-side/new-side view +/// across a drain-then-re-admit either: a `topology_epoch` is bumped on every +/// side pin / un-pin, and `snapshot` retries its materialization if the epoch +/// moves under it (see there). #[derive(Debug)] pub struct PriceLevel { /// The price of this level @@ -52,8 +108,19 @@ pub struct PriceLevel { /// Total hidden quantity at this price level hidden_quantity: AtomicU64, - /// Number of orders at this price level - order_count: AtomicUsize, + /// Packed `(pinned side, resting order count)` — the atomic topology word + /// (issue #126). The side tag lives in the high two bits, the count in the + /// low [`topology::COUNT_BITS`]; see the [`topology`] module for the layout + /// and [`Self::topology_admit`] / [`Self::topology_release_one`] for the CAS + /// protocol. Replaces the former standalone `order_count` counter — the + /// count is now read back out of this word. + topology: AtomicU64, + + /// Monotonic counter bumped on every side pin / un-pin (issue #126). A + /// [`Self::snapshot`] reads it before and after materializing the orders and + /// retries if it moved, so a checksummed snapshot can never capture a torn + /// old-side/new-side view across a drain-then-re-admit transition. + topology_epoch: AtomicU64, /// Queue of orders at this price level orders: OrderQueue, @@ -102,7 +169,7 @@ impl PriceLevel { // violates either would reconstruct a level that trades at the wrong // price or emits contradictory taker sides, so reject it here rather // than restore an incoherent level. - { + let level_side: Option = { let level_price = snapshot.price().as_u128(); let mut level_side = None; for order in snapshot.orders() { @@ -127,7 +194,8 @@ impl PriceLevel { Some(_) => {} } } - } + level_side + }; let order_count = snapshot.orders().len(); let visible_quantity = snapshot.visible_quantity().as_u64(); @@ -137,11 +205,19 @@ impl PriceLevel { let stats = (*snapshot.statistics()).clone(); let queue = OrderQueue::from(snapshot.into_orders()); + // Pin the restored side alongside the restored count in the topology word + // (issue #126). An empty snapshot restores Unpinned; a non-empty one pins + // the single side the validation above proved coherent. `order_count` is + // bounded by the snapshot's own vector length, which fits `COUNT_MASK`. + let side_tag = level_side.map_or(topology::TAG_UNPINNED, topology::tag_of); + let topology_word = topology::pack(side_tag, order_count as u64); + Ok(Self { price, visible_quantity: AtomicU64::new(visible_quantity), hidden_quantity: AtomicU64::new(hidden_quantity), - order_count: AtomicUsize::new(order_count), + topology: AtomicU64::new(topology_word), + topology_epoch: AtomicU64::new(0), orders: queue, stats: Arc::new(stats), }) @@ -192,7 +268,9 @@ impl PriceLevel { price, visible_quantity: AtomicU64::new(0), hidden_quantity: AtomicU64::new(0), - order_count: AtomicUsize::new(0), + // Unpinned side, zero resting orders. + topology: AtomicU64::new(topology::pack(topology::TAG_UNPINNED, 0)), + topology_epoch: AtomicU64::new(0), orders: OrderQueue::new(), stats: Arc::new(PriceLevelStatistics::new()), } @@ -263,9 +341,139 @@ impl PriceLevel { /// [`Self::visible_quantity`]; use [`Self::snapshot`] for a consistent view. #[must_use] pub fn order_count(&self) -> usize { - // `Relaxed`: advisory counter, no happens-before rides on it — see - // `visible_quantity` for the full rationale. - self.order_count.load(Ordering::Relaxed) + // `Relaxed`: advisory read of the count half of the topology word, no + // happens-before rides on it — see `visible_quantity` for the rationale. + topology::count(self.topology.load(Ordering::Relaxed)) as usize + } + + /// The side currently pinned at this level, or `None` if the level is empty + /// (Unpinned). Advisory: a concurrent admission / drain can change it right + /// after the read. + #[must_use] + fn pinned_side(&self) -> Option { + topology::side_of_tag(topology::tag(self.topology.load(Ordering::Relaxed))) + } + + /// Reserve one admission slot for a `side` order: pin the side (or verify it + /// matches the pinned side) and increment the resting-order count, in a + /// single compare-exchange (issue #126). + /// + /// Returns `Ok(true)` iff this call pinned a previously-empty level (the + /// caller then bumps [`Self::topology_epoch`]), `Ok(false)` if it joined an + /// already-pinned same-side level. + /// + /// Because the side and the count move together, two opposite-side + /// admissions into an empty level serialize here: exactly one wins the CAS + /// that pins the side, and the loser then observes a non-empty opposite-side + /// level and is rejected. There is no separate "is the level empty" atomic to + /// fall out of step with the side. + /// + /// # Errors + /// + /// [`PriceLevelError::InvalidOperation`] if `side` is incompatible with the + /// pinned side of a non-empty level, or if the count would exceed + /// [`topology::COUNT_MASK`]. + fn topology_admit(&self, side: Side) -> Result { + let my_tag = topology::tag_of(side); + loop { + let cur = self.topology.load(Ordering::Acquire); + let tag = topology::tag(cur); + let count = topology::count(cur); + if count == 0 { + // Empty level: establish this side with count 1. + let next = topology::pack(my_tag, 1); + if self + .topology + .compare_exchange_weak(cur, next, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + return Ok(true); + } + } else if tag == my_tag { + // Same side: bump the count (checked — never wraps). + let Some(new_count) = count.checked_add(1).filter(|c| *c <= topology::COUNT_MASK) + else { + return Err(PriceLevelError::InvalidOperation { + message: "price level order count overflow on admission".to_string(), + }); + }; + let next = topology::pack(my_tag, new_count); + if self + .topology + .compare_exchange_weak(cur, next, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + return Ok(false); + } + } else { + // Non-empty level pinned to the opposite side: reject. + let resting = topology::side_of_tag(tag); + return Err(PriceLevelError::InvalidOperation { + message: format!( + "order side {side:?} is incompatible with the level's resting side {resting:?}" + ), + }); + } + // Lost the CAS to a concurrent mutation; reload and retry. + } + } + + /// Release one admission slot after removing an order: decrement the count + /// and un-pin the side when it reaches zero, in a single compare-exchange + /// (issue #126). + /// + /// Returns `true` iff this call brought the count to zero and un-pinned the + /// level (the caller then bumps [`Self::topology_epoch`]). Because the un-pin + /// rides the same CAS as the decrement, a concurrent admission either sees + /// the still-pinned non-empty level (and joins / is rejected) or the drained + /// Unpinned level (and establishes) — never an inconsistent in-between. + fn topology_release_one(&self) -> bool { + loop { + let cur = self.topology.load(Ordering::Acquire); + let count = topology::count(cur); + if count == 0 { + // A removal only runs for an order this level held, so the count + // is >= 1; never wrap (crate rule). Treat an impossible underflow + // as a no-op rather than corrupt the word. + debug_assert!(false, "topology count underflow on release"); + return false; + } + let new_count = count - 1; + let next = if new_count == 0 { + topology::pack(topology::TAG_UNPINNED, 0) + } else { + topology::pack(topology::tag(cur), new_count) + }; + if self + .topology + .compare_exchange_weak(cur, next, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + return new_count == 0; + } + } + } + + /// Bump the topology epoch on a side pin / un-pin so a racing + /// [`Self::snapshot`] retries a materialization that spanned the transition. + #[inline] + fn bump_topology_epoch(&self) { + self.topology_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). + fn is_single_side(orders: &[Arc>]) -> bool { + let mut side = None; + for order in orders { + match side { + None => side = Some(order.side()), + Some(s) if s != order.side() => return false, + Some(_) => {} + } + } + true } /// Get the statistics for this price level @@ -303,13 +511,15 @@ impl PriceLevel { /// decided before any counter is touched**: an admission that both reuses a /// live id and would overflow a counter reports /// [`PriceLevelError::DuplicateOrderId`], never the overflow. Only for a - /// free id is capacity reserved, visible → hidden → count, each with a - /// checked [`AtomicU64::fetch_update`] / [`AtomicUsize::fetch_update`] - /// (`checked_add`) — an atomic compare-exchange loop, free of the - /// check-then-`fetch_add` TOCTOU race two admissions near `u64::MAX` would - /// hit. If a later reservation overflows, the earlier ones are rolled back - /// (by the exact delta this call added — a commutative, concurrency-safe - /// undo on these advisory counters), so `try_push_with` publishes nothing + /// free id is capacity reserved: the visible and hidden quantities with a + /// checked [`AtomicU64::fetch_update`] (`checked_add`) — an atomic + /// compare-exchange loop, free of the check-then-`fetch_add` TOCTOU race two + /// admissions near `u64::MAX` would hit — and THEN the side pin and order + /// count together in one compare-exchange (`topology_admit`), which also + /// serializes concurrent opposite-side admissions. If the quantity + /// reservations or the pin fail, the earlier ones are rolled back (by the + /// exact delta this call added — a commutative, concurrency-safe undo on + /// these advisory counters), so `try_push_with` publishes nothing /// and no counter is left drifted. /// /// # Topology invariants @@ -346,20 +556,21 @@ impl PriceLevel { ), }); } - // The level's side is defined by its resting makers: the first maker - // pins it, and a drained (empty) level accepts either side again — the - // queue is the source of truth, so no separate side state has to be - // stored or reset. Any resting order's side is representative (this very - // invariant keeps them all equal). Deriving it is lock-free; see the - // type-level note on the one residual window (two opposite-side - // admissions racing into a genuinely empty level). - if let Some(resting_side) = self.orders.iter_orders().next().map(|o| o.side()) - && order.side() != resting_side + // The level's side is pinned in the topology word (issue #126): the + // first maker pins it, later same-side makers join, and the drain that + // empties the level un-pins it so a drained level accepts either side + // again. This is a cheap EARLY reject of an opposite-side order against a + // non-empty level, so the common mismatch never reserves counter + // capacity. It is only an optimization — the AUTHORITATIVE, race-free + // side decision is the pin CAS ([`Self::topology_admit`]) run inside the + // reservation closure below, which serializes concurrent admissions. + let order_side = order.side(); + if let Some(resting_side) = self.pinned_side() + && order_side != resting_side { return Err(PriceLevelError::InvalidOperation { message: format!( - "order side {:?} is incompatible with the level's resting side {resting_side:?}", - order.side() + "order side {order_side:?} is incompatible with the level's resting side {resting_side:?}" ), }); } @@ -399,15 +610,17 @@ impl PriceLevel { // lock, so a concurrent cancel + readmission can never split this id // across two index entries. // - // Inside the closure, capacity is reserved visible → hidden → count with - // checked `fetch_update` (an atomic CAS loop, free of the - // check-then-`fetch_add` TOCTOU race two admissions near `u64::MAX` - // would hit). `Relaxed` throughout: these are the advisory counters - // (issue #68); the queue publication carries the real happens-before, - // not these RMWs. If a later reservation overflows, the earlier ones are - // rolled back by the exact delta this call added — a commutative, - // concurrency-safe undo — before returning `Err`, so `try_push_with` - // publishes nothing and no counter is left drifted. + // Inside the closure, capacity is reserved visible → hidden with checked + // `fetch_update` (an atomic CAS loop, free of the check-then-`fetch_add` + // TOCTOU race two admissions near `u64::MAX` would hit), and THEN the + // side is pinned and the count bumped in one CAS via `topology_admit`. + // `Relaxed` on the advisory visible / hidden RMWs (issue #68); the pin + // CAS uses `AcqRel` because side coherence is a hard invariant, not an + // advisory counter. The pin goes LAST so it is only mutated on the + // success path: an incompatible side or a count overflow returns `Err` + // after rolling back the visible + hidden reservations this call made + // (a commutative, concurrency-safe undo), leaving the topology word + // untouched and `try_push_with` publishing nothing. let order_arc = Arc::new(order); self.orders.try_push_with(order_arc.clone(), || { if self @@ -437,19 +650,28 @@ impl PriceLevel { }); } - if self - .order_count - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| c.checked_add(1)) - .is_err() - { - // Roll back the visible + hidden reservations this call made. - self.visible_quantity - .fetch_sub(visible_qty, Ordering::Relaxed); - self.hidden_quantity - .fetch_sub(hidden_qty, Ordering::Relaxed); - return Err(PriceLevelError::InvalidOperation { - message: "price level order count overflow on admission".to_string(), - }); + // Pin the side and bump the count in one CAS. This is the + // authoritative, race-free side decision: two opposite-side + // admissions into an empty level serialize here, only one wins. + match self.topology_admit(order_side) { + Ok(established) => { + if established { + // Pinned a previously-empty level; bump the epoch BEFORE + // the publish that `try_push_with` does next, so a + // snapshot whose walk spans this transition sees the epoch + // move and retries. + self.bump_topology_epoch(); + } + } + Err(err) => { + // Roll back the visible + hidden reservations this call made; + // the topology word was not mutated (pin goes last). + self.visible_quantity + .fetch_sub(visible_qty, Ordering::Relaxed); + self.hidden_quantity + .fetch_sub(hidden_qty, Ordering::Relaxed); + return Err(err); + } } Ok(()) @@ -743,6 +965,33 @@ impl PriceLevel { timestamp: TimestampMs, trade_id_generator: &UuidGenerator, ) -> MatchResult { + // -------- Self-match is terminal (issue #126, tightening #120) -------- + // + // If the taker's own id already rests at this level, the taker cannot + // take liquidity here: matching would either self-trade (forbidden) or, + // via the in-sweep skip, walk PAST its own resting order to trade with + // OTHER makers — but issue #120's acceptance is that a self-match attempt + // emits NO trades and leaves the level byte-identical. So reject + // terminally, before any sweep, for EVERY TIF and kind. This check + // precedes and therefore dominates the post-only / fill-or-kill + // pre-checks below (a self-match `Fok` is Rejected, not Killed); the + // post-only behaviour for a taker that does NOT rest here is unchanged. + // The lookup is an O(1) id probe. The in-sweep `SelfTradeSkipped` path is + // retained as documented defense-in-depth for the narrow race where the + // taker's resting order is admitted AFTER this probe but during the + // sweep — even then it must never self-trade. + if incoming_quantity > 0 && self.orders.find(taker_order_id).is_some() { + tracing::debug!( + taker_order_id = %taker_order_id, + incoming_quantity, + price = self.price, + "taker rejected: own order id already rests at this level (self-match)" + ); + let mut result = MatchResult::new(taker_order_id, Quantity::new(incoming_quantity)); + result.mark_rejected(incoming_quantity); + return result; + } + // -------- Taker TIF / kind pre-checks (before any queue mutation) -------- // // PostOnly: must never take liquidity. If any matchable depth exists for @@ -871,9 +1120,14 @@ impl PriceLevel { while remaining > 0 { let outcome = self.orders.match_front(&mut set_aside, |seq, order_arc| { - // Self-trade prevention (deterministic in every build profile, - // not a debug-only assert): a resting maker must never trade - // against a taker carrying the same id. Skip it — park its + // Self-trade prevention, DEFENSE-IN-DEPTH (issue #126). The + // common case is already handled terminally before the sweep: if + // the taker id rests here, `match_order` returns `Rejected` with + // no trades. This in-sweep skip covers only the narrow race where + // the taker's own order is admitted AFTER that pre-check but + // before the sweep reaches its slot. Deterministic in every build + // profile (not a debug-only assert): a resting maker must never + // trade against a taker carrying the same id. Skip it — park its // sequence like a no-progress maker so the sweep advances to the // makers behind it — rather than emit a self-trade. The maker is // left untouched (no trade, counters and queue unchanged). @@ -1100,7 +1354,11 @@ impl PriceLevel { if data.fully_consumed { // Maker fully consumed and removed inside `match_front`. - self.order_count.fetch_sub(1, Ordering::Relaxed); + // Decrement the count and un-pin if this drained the level + // (issue #126); the removal already happened-before here. + if self.topology_release_one() { + self.bump_topology_epoch(); + } if data.hidden_stranded > 0 { self.hidden_quantity .fetch_sub(data.hidden_stranded, Ordering::Relaxed); @@ -1155,7 +1413,26 @@ impl PriceLevel { // sequence) order so a snapshot round-trip re-enqueues them in identical // priority order; every aggregate is derived from this same snapshot so // they are mutually consistent by construction. - let orders = self.snapshot_by_insertion_seq(); + // + // Guard against a TORN topology (issue #126): a walk that spans a + // drain-then-re-admit to the opposite side could capture old-side and + // new-side orders together, producing a checksummed snapshot that + // `from_snapshot` would reject for mixed sides. `topology_epoch` is + // bumped on every side pin / un-pin, so if it moves across the walk we + // retry. The `is_single_side` fallback GUARANTEES termination: a stable + // epoch already implies a coherent walk (a tear requires a transition, + // which bumps the epoch), so we only ever loop while a walk actually came + // back mixed-side, which needs an in-progress opposite-side flip — once + // flipping stops (finite writers) the next walk is coherent and returns. + let orders = loop { + let epoch_before = self.topology_epoch.load(Ordering::Acquire); + let orders = self.snapshot_by_insertion_seq(); + let epoch_after = self.topology_epoch.load(Ordering::Acquire); + if epoch_before == epoch_after || Self::is_single_side(&orders) { + break orders; + } + // A side transition raced the walk AND left a mixed-side view; retry. + }; let order_count = orders.len(); @@ -1289,7 +1566,11 @@ impl PriceLevel { .fetch_sub(visible_qty, Ordering::Relaxed); self.hidden_quantity .fetch_sub(hidden_qty, Ordering::Relaxed); - self.order_count.fetch_sub(1, Ordering::Relaxed); + // Decrement the count and un-pin if this drained the + // level (issue #126); the `remove` above happened-before. + if self.topology_release_one() { + self.bump_topology_epoch(); + } // Update statistics self.stats.record_order_removed(); @@ -1443,7 +1724,11 @@ impl PriceLevel { .fetch_sub(visible_qty, Ordering::Relaxed); self.hidden_quantity .fetch_sub(hidden_qty, Ordering::Relaxed); - self.order_count.fetch_sub(1, Ordering::Relaxed); + // Decrement the count and un-pin if this drained the + // level (issue #126); the `remove` above happened-before. + if self.topology_release_one() { + self.bump_topology_epoch(); + } // Update statistics self.stats.record_order_removed(); @@ -1474,7 +1759,11 @@ impl PriceLevel { .fetch_sub(visible_qty, Ordering::Relaxed); self.hidden_quantity .fetch_sub(hidden_qty, Ordering::Relaxed); - self.order_count.fetch_sub(1, Ordering::Relaxed); + // Decrement the count and un-pin if this drained the level + // (issue #126); the `remove` above happened-before. + if self.topology_release_one() { + self.bump_topology_epoch(); + } // Update statistics self.stats.record_order_removed(); @@ -1506,7 +1795,11 @@ impl PriceLevel { .fetch_sub(visible_qty, Ordering::Relaxed); self.hidden_quantity .fetch_sub(hidden_qty, Ordering::Relaxed); - self.order_count.fetch_sub(1, Ordering::Relaxed); + // Decrement the count and un-pin if this drained the + // level (issue #126); the `remove` above happened-before. + if self.topology_release_one() { + self.bump_topology_epoch(); + } // Update statistics self.stats.record_order_removed(); diff --git a/src/price_level/order_queue.rs b/src/price_level/order_queue.rs index 5a4c51d..e949bc2 100644 --- a/src/price_level/order_queue.rs +++ b/src/price_level/order_queue.rs @@ -333,6 +333,18 @@ impl OrderQueue { // escapes into a `FrontAction`). let (action, result) = decide(seq, occupied.get().1.as_ref()); + // A `SetAside` records a sequence into the caller's scratch + // `HashSet`, whose first insert allocates. Defer that insert + // until AFTER the entry lock is released (issue #126) so no + // allocation ever runs under the shard lock — the set is + // per-sweep scratch owned by the caller, never shared, so it + // needs no lock protection. The other actions commit their + // queue mutations here, under the lock, as before. + let mut park_seq: Option = None; + // Every arm releases the entry lock by the time it finishes + // (either `occupied.remove()` consumes it, or an explicit + // `drop`), so the deferred `set_aside` insert below never runs + // under the shard lock. match &action { FrontAction::Remove => { // Full consume: remove the entry under the lock, then @@ -345,8 +357,8 @@ impl OrderQueue { // Partial fill keeping priority: swap the stored value // to the residual in place, keeping the same // sequence/index entry. Still under the entry lock. - let slot = occupied.get_mut(); - slot.1 = residual.clone(); + occupied.get_mut().1 = residual.clone(); + drop(occupied); } FrontAction::ReplaceAtTail(refreshed) => { // Replenished tranche loses time priority, but the @@ -362,13 +374,12 @@ impl OrderQueue { slot.0 = new_seq; slot.1 = refreshed.clone(); } - // `occupied` still holds the per-entry lock here (it - // is dropped at the end of this arm), so re-keying the - // index — a different structure (`SkipMap`), no - // deadlock — happens while a concurrent cancel is - // still excluded from the entry. Once the lock is - // released the value already carries `new_seq`, so a - // cancel removes `orders[id]` and `index[new_seq]` + // `occupied` still holds the per-entry lock here, so + // re-keying the index — a different structure + // (`SkipMap`), no deadlock — happens while a concurrent + // cancel is still excluded from the entry. Once the + // lock is released the value already carries `new_seq`, + // so a cancel removes `orders[id]` and `index[new_seq]` // consistently. The only residue a race can leave is a // stale `index[seq|new_seq] -> id` entry pointing at an // already-removed id, which the next `match_front` @@ -376,14 +387,22 @@ impl OrderQueue { // counter update is ever lost. self.index.remove(&seq); self.index.insert(new_seq, order_id); + drop(occupied); } FrontAction::SetAside => { - // No progress: leave the entry untouched and park its - // sequence so the sweep advances past it. - set_aside.insert(seq); + // No progress: leave the entry untouched. Release the + // lock and park its sequence below. + drop(occupied); + park_seq = Some(seq); } } + // The entry lock is released on every arm above; a + // possibly-allocating scratch-set insert now runs unlocked. + if let Some(seq) = park_seq { + set_aside.insert(seq); + } + return FrontOutcome::Matched { result }; } } diff --git a/src/price_level/tests/level.rs b/src/price_level/tests/level.rs index 74d26b7..a08f7e0 100644 --- a/src/price_level/tests/level.rs +++ b/src/price_level/tests/level.rs @@ -7025,62 +7025,81 @@ mod tests { } #[test] - fn test_match_order_self_trade_skipped() { - // Makers 1, 2, 3 rest in FIFO order; the taker shares maker 1's id. - let level = PriceLevel::new(10_000); - level - .add_order(create_standard_order(1, 10_000, 40)) - .expect("maker 1 admits"); - level - .add_order(create_standard_order(2, 10_000, 30)) - .expect("maker 2 admits"); - level - .add_order(create_standard_order(3, 10_000, 50)) - .expect("maker 3 admits"); - + fn test_match_order_self_match_terminal_rejected_all_tifs() { + // Issue #126: a self-match is TERMINAL. If the taker's own id rests at + // the level, the match emits NO trades and leaves the level + // byte-identical for EVERY TIF and kind — it does NOT walk past its own + // resting order to trade with the other makers (the old skip behaviour). let namespace = Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap(); - let generator = UuidGenerator::new(namespace); - // Taker id == maker 1's id: maker 1 must be skipped, 2 and 3 consumed. - let result = level.match_order( - 1_000, - Id::from_u64(1), - TimeInForce::Gtc, - TakerKind::Standard, - TimestampMs::new(1_700_000_000_000), - &generator, - ); - // No trade names the taker as its own maker (no self-trade emitted). - let makers: Vec = result - .trades() - .as_vec() - .iter() - .map(|t| t.maker_order_id()) - .collect(); - assert!( - makers.iter().all(|m| *m != Id::from_u64(1)), - "no trade may have maker == taker; got {makers:?}" - ); - // The other makers are still consumed, in FIFO order. - assert_eq!(makers, vec![Id::from_u64(2), Id::from_u64(3)]); - // MatchResult stays consistent: executed == 30 + 50 = 80. - assert_eq!( - result.executed_quantity().expect("no overflow").as_u64(), - 80 - ); - // The skipped maker 1 is left resting, untouched. - let resting: Vec = level - .snapshot_by_insertion_seq() - .iter() - .map(|o| o.id()) - .collect(); - assert_eq!(resting, vec![Id::from_u64(1)]); + // Every TIF, plus the post-only kind, must reject identically. + let cases: [(TimeInForce, TakerKind); 6] = [ + (TimeInForce::Gtc, TakerKind::Standard), + (TimeInForce::Ioc, TakerKind::Standard), + (TimeInForce::Fok, TakerKind::Standard), + (TimeInForce::Day, TakerKind::Standard), + (TimeInForce::Gtc, TakerKind::PostOnly), + (TimeInForce::Fok, TakerKind::PostOnly), + ]; + + for (tif, kind) in cases { + // Makers 1, 2, 3 rest in FIFO order; the taker shares maker 1's id. + let level = PriceLevel::new(10_000); + level + .add_order(create_standard_order(1, 10_000, 40)) + .expect("maker 1 admits"); + level + .add_order(create_standard_order(2, 10_000, 30)) + .expect("maker 2 admits"); + level + .add_order(create_standard_order(3, 10_000, 50)) + .expect("maker 3 admits"); + let before = level.snapshot_by_insertion_seq(); + + let result = level.match_order( + 1_000, + Id::from_u64(1), + tif, + kind, + TimestampMs::new(1_700_000_000_000), + &UuidGenerator::new(namespace), + ); + + // Terminal Rejected: zero trades, full remaining, nothing executed. + assert!( + result.was_rejected(), + "self-match must be Rejected for {tif:?}/{kind:?}" + ); + assert_eq!(result.trades().len(), 0, "{tif:?}/{kind:?}: no trades"); + assert_eq!( + result.remaining_quantity().as_u64(), + 1_000, + "{tif:?}/{kind:?}: full remaining" + ); + assert_eq!( + result.executed_quantity().expect("no overflow").as_u64(), + 0, + "{tif:?}/{kind:?}: nothing executed" + ); + + // The level is 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::>(), + "{tif:?}/{kind:?}: queue unchanged" + ); + assert_eq!(level.order_count(), 3, "{tif:?}/{kind:?}: count unchanged"); + assert_counters_match_queue(&level); + } } #[test] - fn test_matchable_quantity_self_trade_parity_with_fok() { - // Maker 1 (shares the taker id) has 40; maker 2 has 60. A taker id 1 - // can only take maker 2's 60 (maker 1 is self-trade-skipped). + fn test_matchable_quantity_self_skip_but_match_order_rejects_self() { + // Maker 1 (shares the taker id) has 40; maker 2 has 60. The dry-run + // helper `matchable_quantity` still skips the self-trade maker (it backs + // the in-sweep defense-in-depth path, where the taker's order is admitted + // mid-sweep), so it reports 60 takeable. let level = PriceLevel::new(10_000); level .add_order(create_standard_order(1, 10_000, 40)) @@ -7089,48 +7108,203 @@ mod tests { .add_order(create_standard_order(2, 10_000, 60)) .expect("maker 2 admits"); - // The dry run agrees: skipping the self-trade maker leaves 60. assert_eq!(level.matchable_quantity(100, Id::from_u64(1)), 60); assert_eq!(level.matchable_quantity(60, Id::from_u64(1)), 60); + // But `match_order` is TERMINAL when the taker id already rests (issue + // #126): it rejects up front for every TIF, dominating the FOK dry run — + // a self-match FOK is Rejected, NOT killed and NOT filled from maker 2. let namespace = Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap(); - // A FOK taker (id 1) of 100 must be KILLED: only 60 is takeable. - let killed = level.match_order( - 100, - Id::from_u64(1), - TimeInForce::Fok, - TakerKind::Standard, - TimestampMs::new(1_700_000_000_000), - &UuidGenerator::new(namespace), - ); - assert!( - killed.was_killed(), - "FOK 100 must be killed (only 60 takeable)" - ); - assert_eq!(killed.trades().len(), 0); - assert_eq!( - level.order_count(), - 2, - "a killed FOK leaves the queue untouched" - ); + for qty in [100u64, 60] { + let result = level.match_order( + qty, + Id::from_u64(1), + TimeInForce::Fok, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &UuidGenerator::new(namespace), + ); + assert!( + result.was_rejected(), + "self-match FOK({qty}) is Rejected, not killed/filled" + ); + assert!(!result.was_killed(), "self-match is Rejected, not Killed"); + assert_eq!(result.trades().len(), 0, "no trades on self-match"); + assert_eq!(result.remaining_quantity().as_u64(), qty); + assert_eq!( + level.order_count(), + 2, + "a rejected self-match leaves the queue untouched" + ); + } - // A FOK taker (id 1) of 60 must FILL: exactly maker 2's 60. + // A taker with a DISTINCT id (id 3) does take maker 1 + maker 2 normally. let filled = level.match_order( - 60, - Id::from_u64(1), + 100, + Id::from_u64(3), TimeInForce::Fok, TakerKind::Standard, TimestampMs::new(1_700_000_000_001), &UuidGenerator::new(namespace), ); - assert!(filled.is_complete(), "FOK 60 must fill"); + assert!(filled.is_complete(), "non-self FOK 100 fills 40 + 60"); let makers: Vec = filled .trades() .as_vec() .iter() .map(|t| t.maker_order_id()) .collect(); - assert_eq!(makers, vec![Id::from_u64(2)], "only maker 2 fills the FOK"); + assert_eq!(makers, vec![Id::from_u64(1), Id::from_u64(2)]); + } + + #[test] + fn test_opposite_side_admissions_race_exactly_one_wins() { + // Issue #126 🔴: two opposite-side admissions racing into a genuinely + // empty level must NEVER both admit — the atomic side pin serializes + // them so exactly one wins and the other is rejected. Under the old + // derive-from-queue scheme both could observe an empty queue and admit. + use std::sync::{Arc, Barrier}; + use std::thread; + + const ITERATIONS: usize = 3_000; + const PRICE: u128 = 10_000; + + for iter in 0..ITERATIONS { + let level = Arc::new(PriceLevel::new(PRICE)); + let barrier = Arc::new(Barrier::new(2)); + let buy_id = iter as u64 * 2 + 1; + let sell_id = buy_id + 1; + + let buyer = { + let level = Arc::clone(&level); + let barrier = Arc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + level.add_order(create_standard_order(buy_id, PRICE, 10)) + }) + }; + let seller = { + let level = Arc::clone(&level); + let barrier = Arc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + level.add_order(create_sell_standard_order(sell_id, PRICE, 10)) + }) + }; + + let buy_res = buyer.join().expect("buyer thread panicked"); + let sell_res = seller.join().expect("seller thread panicked"); + + // Exactly one admission wins. + let admitted = usize::from(buy_res.is_ok()) + usize::from(sell_res.is_ok()); + assert_eq!( + admitted, + 1, + "iter {iter}: exactly one opposite-side admission may win (buy_ok={}, sell_ok={})", + buy_res.is_ok(), + sell_res.is_ok() + ); + // The loser is rejected with an incompatible-side error. + if let Err(err) = &buy_res { + assert!(matches!(err, PriceLevelError::InvalidOperation { .. })); + } + if let Err(err) = &sell_res { + assert!(matches!(err, PriceLevelError::InvalidOperation { .. })); + } + + // The level holds exactly one order; snapshot is single-side; the + // advisory counters agree with the queue. + assert_eq!(level.order_count(), 1, "iter {iter}"); + let snap = level.snapshot(); + assert_eq!(snap.orders().len(), 1, "iter {iter}"); + assert_counters_match_queue(&level); + + // A drained level re-accepts EITHER side (the pin un-pinned on drain). + let winner_id = if buy_res.is_ok() { buy_id } else { sell_id }; + level + .update_order(OrderUpdate::Cancel { + order_id: Id::from_u64(winner_id), + }) + .expect("cancel winner") + .expect("winner was resting"); + assert_eq!(level.order_count(), 0, "iter {iter}: drained"); + // Whichever side lost the race can now be admitted into the empty level. + let readmit = if buy_res.is_ok() { + level.add_order(create_sell_standard_order(sell_id, PRICE, 7)) + } else { + level.add_order(create_standard_order(buy_id, PRICE, 7)) + }; + assert!( + readmit.is_ok(), + "iter {iter}: a drained level must re-accept the opposite side" + ); + } + } + + #[test] + fn test_snapshot_never_captures_torn_side_under_flips() { + // Issue #126 🔴: a snapshot walk that spans a drain-then-re-admit to the + // opposite side must never capture a torn old-side/new-side view (which + // `from_snapshot` would reject for mixed sides). The topology epoch makes + // `snapshot` retry across such a transition; here a flipper thread churns + // the level Buy-batch -> drained -> Sell-batch -> drained while the main + // thread takes many snapshots and asserts each is single-side. + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::thread; + + const PRICE: u128 = 10_000; + const BATCH: u64 = 8; + + let level = Arc::new(PriceLevel::new(PRICE)); + let done = Arc::new(AtomicBool::new(false)); + + let flipper = { + let level = Arc::clone(&level); + let done = Arc::clone(&done); + thread::spawn(move || { + let mut round = 0u64; + while !done.load(Ordering::Relaxed) { + let buy = round.is_multiple_of(2); + let base = 1_000 + round * BATCH; + for i in 0..BATCH { + let id = base + i; + let order = if buy { + create_standard_order(id, PRICE, 5) + } else { + create_sell_standard_order(id, PRICE, 5) + }; + // May transiently fail if the opposite side is still + // draining; that is fine, we just churn the topology. + let _ = level.add_order(order); + } + for i in 0..BATCH { + let _ = level.update_order(OrderUpdate::Cancel { + order_id: Id::from_u64(base + i), + }); + } + round += 1; + } + }) + }; + + for _ in 0..50_000 { + let snap = level.snapshot(); + let mut side = None; + for order in snap.orders() { + match side { + None => side = Some(order.side()), + Some(s) => assert_eq!( + s, + order.side(), + "snapshot captured a torn mixed-side view (issue #126)" + ), + } + } + } + + done.store(true, Ordering::Relaxed); + flipper.join().expect("flipper thread panicked"); } #[test] From 9edab14ea8997032d058064aaf1f7881e55537a7 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 12:57:46 +0200 Subject: [PATCH 3/4] fix(price_level)!: resequence quantity increases without vacating the id The quantity-increase branch demoted via remove(id) + push(new): during that gap the id was absent from the queue map, so a concurrent cancel reported no removal while the update re-inserted the maker (lost cancel / resurrection), a concurrent same-id admission could win try_push and then be blindly overwritten (the P1 flagged in the #113 audit), and the transient absence was the second side-topology race window documented in #120. match_front also never verified that the index-selected sequence still owned the live map entry, so a stale front selection could act on a just-demoted maker. New OrderQueue::resequence_to_tail extends the ReplaceAtTail shape: under one continuous hold of the DashMap entry lock it mints a tail sequence, swaps the stored (seq, order) pair in place, and re-keys the index -- the id never leaves the map. A concurrent cancel now either fully precedes (update reports not-found) or fully follows (it removes the re-sequenced order); a same-id admission is always rejected as DuplicateOrderId. match_front gains a stale-seq guard: if the stored sequence no longer equals the index-selected one, the stale key is dropped and the scan retries -- sequences are monotonic and never reused, so the removal is provably safe. Demotion semantics unchanged (fresh tail seq, original timestamp; #109 tests untouched). BREAKING: OrderQueue::push (blind overwrite) has no production caller left and is retired to #[cfg(test)] -- admission uses try_push, demotion uses resequence_to_tail. Migration guide + README updated; the now-closed #113/#120 doc caveats are retired. Barrier race tests (1500 iterations each): upsize-vs-cancel never loses a cancel or resurrects; upsize-vs-duplicate-admission is always rejected with no counter drift (fails 3/3 under the old remove+push); concurrent upsize+match stress drains consistently. Perf: update-heavy bench 26.0us vs 25.7us median -- neutral within noise; the win is correctness. Closes #119 --- README.md | 19 +++ src/lib.rs | 19 +++ src/price_level/level.rs | 30 ++-- src/price_level/order_queue.rs | 97 +++++++++++-- src/price_level/tests/level.rs | 248 ++++++++++++++++++++++++++++++++- 5 files changed, 390 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index 5f4d808..2ff0488 100644 --- a/README.md +++ b/README.md @@ -581,6 +581,25 @@ order ids owned by the same `user_id` will still trade. Account-level STP is the responsibility of the order book composing these levels, which owns the account relationships a single price level does not. +### Migration Guide (atomic quantity-increase re-sequencing) + +A quantity increase via [`PriceLevel::update_order`] still demotes the maker +to the back of the queue (fresh tail sequence, original timestamp), but it +now does so **in place** — the order id never leaves the internal map. This +closes the concurrency window the previous `remove` + re-insert opened +(issue #119): a concurrent cancel can no longer be lost or resurrect the +order, a concurrent same-id admission can no longer slip into the gap +(`add_order` for a live id is always rejected), and the match sweep can no +longer act on a stale front position. The public behaviour of `update_order` +is unchanged; only its concurrency safety improves. + +The internal `OrderQueue::push` — a blind, overwrite-on-collision insert +with no remaining production caller — is removed from the public API (it is +now test-only). Admission uses `try_push` (insert-if-absent) and the +quantity-increase demotion uses the internal atomic re-sequence, so `push` +was a footgun with no safe use; construct queues through [`PriceLevel`]'s +public surface instead. + ## Setup Instructions diff --git a/src/lib.rs b/src/lib.rs index e39dd6b..c6e9dfd 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -573,6 +573,25 @@ //! the responsibility of the order book composing these levels, which owns the //! account relationships a single price level does not. //! +//! ## Migration Guide (atomic quantity-increase re-sequencing) +//! +//! A quantity increase via [`PriceLevel::update_order`] still demotes the maker +//! to the back of the queue (fresh tail sequence, original timestamp), but it +//! now does so **in place** — the order id never leaves the internal map. This +//! closes the concurrency window the previous `remove` + re-insert opened +//! (issue #119): a concurrent cancel can no longer be lost or resurrect the +//! order, a concurrent same-id admission can no longer slip into the gap +//! (`add_order` for a live id is always rejected), and the match sweep can no +//! longer act on a stale front position. The public behaviour of `update_order` +//! is unchanged; only its concurrency safety improves. +//! +//! The internal `OrderQueue::push` — a blind, overwrite-on-collision insert +//! with no remaining production caller — is removed from the public API (it is +//! now test-only). Admission uses `try_push` (insert-if-absent) and the +//! quantity-increase demotion uses the internal atomic re-sequence, so `push` +//! was a footgun with no safe use; construct queues through [`PriceLevel`]'s +//! public surface instead. +//! mod orders; mod price_level; diff --git a/src/price_level/level.rs b/src/price_level/level.rs index 3372bbe..7a0645c 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -90,8 +90,9 @@ mod topology { /// observes a non-empty opposite-side level and is rejected. /// - An **opposite-side admission racing a same-side upsize** cannot slip /// through a transient queue gap: the pin persists across a maker's -/// remove-then-re-add because the count never reaches zero, so the side stays -/// pinned even while the queue momentarily looks empty. +/// demotion because the count never reaches zero (and as of issue #119 the +/// quantity-increase demotion re-sequences in place without vacating the id +/// at all). /// /// A concurrent [`Self::snapshot`] cannot capture a torn old-side/new-side view /// across a drain-then-re-admit either: a `topology_epoch` is bumped on every @@ -1658,15 +1659,24 @@ impl PriceLevel { // removed/replaced, so the counter deltas reflect the real // transition rather than the possibly-stale pre-read above. let old = if new_total > prev_total { - // Quantity INCREASE: demote to the back of the queue (mint a - // new sequence), losing time priority. Retains the - // remove+push shape; its concurrency window is the broader - // #81 work and is intentionally not addressed here. - let Some(removed) = self.orders.remove(order_id) else { + // Quantity INCREASE: demote to the back of the queue (a fresh + // tail sequence), losing time priority. Uses the atomic + // `resequence_to_tail` primitive (issue #119): the id stays + // resident in the MAP throughout — no absent window — so a + // concurrent cancel cannot be lost, a same-id admission is + // always rejected, and match-front can no longer act on a + // stale front position. (The INDEX is still re-keyed in two + // steps, so one match scan may transiently miss the maker — + // a missed fill at worst, strictly narrower than the old + // remove+push absence.) Returns the replaced order (for the + // counter deltas below) or `None` if concurrently removed. + let Some(replaced) = self + .orders + .resequence_to_tail(order_id, new_order_arc.clone()) + else { return Ok(None); // Removed by another thread. }; - self.orders.push(new_order_arc.clone()); - removed + replaced } else { // Quantity DECREASE or unchanged total: keep the maker's // queue position by swapping the stored order in place at its @@ -1688,7 +1698,7 @@ impl PriceLevel { let old_visible = old.visible_quantity().as_u64(); let old_hidden = old.hidden_quantity().as_u64(); // `Relaxed` on both branches: advisory counters (issue #68); the - // queue mutation above (`update_in_place` / `remove` + `push`) + // queue mutation above (`update_in_place` / `resequence_to_tail`) // carries the happens-before, not these counter RMWs. let apply = |counter: &std::sync::atomic::AtomicU64, old: u64, new: u64| { if new >= old { diff --git a/src/price_level/order_queue.rs b/src/price_level/order_queue.rs index e949bc2..72e5716 100644 --- a/src/price_level/order_queue.rs +++ b/src/price_level/order_queue.rs @@ -91,18 +91,14 @@ impl OrderQueue { /// Add an order to the tail of the queue (newest time priority), /// **unconditionally overwriting** any existing entry for the same id. /// - /// This is the re-insert primitive for a maker that was just removed and is - /// being put back under a fresh sequence (the upsize `remove` + `push` - /// demotion), where the id is guaranteed absent. For *admission*, where the - /// id may collide with a live order, use [`OrderQueue::try_push`], which - /// rejects the duplicate instead of overwriting it (leaving the id-keyed map - /// and the ordered index disagreeing). - /// - /// `pub(crate)`: overwriting publication is never safe to expose — a caller - /// that reused a live id would silently replace the resting order and strand - /// its old index entry. Admission goes through [`OrderQueue::try_push`] / - /// [`OrderQueue::try_push_with`]; this stays available only to the in-crate - /// upsize re-insert, whose id is provably absent at the call site. + /// Test-only queue-building fixture. It has no production caller: admission + /// uses [`OrderQueue::try_push`] / [`OrderQueue::try_push_with`] + /// (insert-if-absent, issue #113) and the quantity-increase demotion uses + /// [`OrderQueue::resequence_to_tail`] (in-place, issue #119). Its blind + /// overwrite would leave the id-keyed map and the ordered index + /// disagreeing, so it is deliberately not part of the public API — like + /// [`OrderQueue::reinsert`], it is `#[cfg(test)]`. + #[cfg(test)] pub(crate) fn push(&self, order: Arc>) { // `Relaxed` is sufficient: only the uniqueness and monotonicity of the // counter matter. The happens-before ordering between concurrent @@ -123,6 +119,9 @@ impl OrderQueue { /// publication has no side effects to commit atomically with it; the /// reservation-hook form commits a caller-side reservation (e.g. the level's /// atomic counters) under the same shard lock that decides the id is free. + /// The quantity-increase update path no longer vacates the id either: as of + /// issue #119 it demotes in place via `resequence_to_tail`, so there is no + /// remove-then-push window for a same-id admission to slip into. /// /// # Errors /// @@ -324,6 +323,26 @@ impl OrderQueue { continue; } Entry::Occupied(mut occupied) => { + // Stale front-selection guard (issue #119). The `(seq, id)` + // pair was read from the index BEFORE this entry lock was + // taken. A concurrent `resequence_to_tail` (quantity-increase + // demotion) may have moved this maker to a fresh tail + // sequence in that gap, so the entry now stores a DIFFERENT + // sequence and the maker is no longer the front. Acting on it + // via the stale front position would break FIFO. Drop the + // stale index key (the demoted maker already lives under its + // new key) and retry with a fresh front read — the same + // self-heal shape as the `Vacant` arm above. Sequences are + // monotonic and never reused, so `index[seq]` can only ever + // have pointed at this id, making the removal safe. The + // retry is unbounded; liveness relies on re-sequencings of + // the front maker being finite (the single-logical-writer + // update contract), as with the `Vacant` self-heal. + if occupied.get().0 != seq { + self.index.remove(&seq); + continue; + } + // `occupied.get()` is `(stored_seq, order)`. Decide against // the live order while the entry lock is held. Borrow the // resident order rather than cloning its `Arc` on the hot @@ -409,6 +428,60 @@ impl OrderQueue { } } + /// Atomically re-sequence `order_id` to the tail (a fresh insertion + /// sequence), swapping in `new_order`, **without ever removing the id from + /// the map**. Returns the replaced order, or `None` if the id was + /// concurrently removed (nothing is inserted in that case). + /// + /// This is the quantity-increase demotion primitive (issue #119). It is the + /// standalone form of the [`FrontAction::ReplaceAtTail`] path the match + /// sweep already uses: it holds the `DashMap` per-entry (shard) lock across + /// the whole operation — mint a tail sequence, swap the stored + /// `(sequence, order)` pair in place, then re-key the index + /// (`old_seq -> new_seq`) — so the id stays continuously resident in + /// `orders`. The prior `remove` + `push` demotion opened an absent window in + /// which the id was gone from the map; a concurrent cancel could report no + /// removal while the update re-inserted (lost cancel / resurrection), a + /// concurrent same-id admission could slip into the gap, and + /// [`OrderQueue::match_front`] could act on a stale front. Because the id + /// never leaves the map here, a concurrent [`OrderQueue::remove`] either + /// fully precedes this call (this returns `None`) or fully follows it (it + /// removes the re-sequenced order): all three hazards are closed. Note the + /// index re-key is still two SkipMap ops, so a concurrent front scan can + /// transiently miss the maker in the INDEX (map residency is unaffected) — + /// at worst a missed fill for that sweep, the same transient + /// [`FrontAction::ReplaceAtTail`] has always had and strictly narrower + /// than the old remove+push map absence. + /// + /// Allocation-free beyond the `Arc` the caller hands in. + pub(crate) fn resequence_to_tail( + &self, + order_id: Id, + new_order: Arc>, + ) -> Option>> { + match self.orders.entry(order_id) { + // Concurrently removed: do not resurrect it. + Entry::Vacant(_) => None, + Entry::Occupied(mut occupied) => { + // Mint the tail sequence and swap the stored (seq, order) pair in + // place under the entry lock, then re-key the index while the + // lock is still held (a different structure — no deadlock), + // exactly as `ReplaceAtTail` does. + let new_seq = self.next_seq.fetch_add(1, Ordering::Relaxed); + let (old_seq, replaced) = { + let slot = occupied.get_mut(); + let old_seq = slot.0; + let replaced = std::mem::replace(&mut slot.1, new_order); + slot.0 = new_seq; + (old_seq, replaced) + }; + self.index.remove(&old_seq); + self.index.insert(new_seq, order_id); + Some(replaced) + } + } + } + /// Re-insert an order at a given (previously assigned) insertion sequence. /// /// Re-inserting at a maker's original sequence returns it to its place in diff --git a/src/price_level/tests/level.rs b/src/price_level/tests/level.rs index a08f7e0..256c135 100644 --- a/src/price_level/tests/level.rs +++ b/src/price_level/tests/level.rs @@ -1,7 +1,7 @@ #[cfg(test)] mod tests { use crate::errors::PriceLevelError; - use crate::execution::{MatchOutcome, TakerKind}; + use crate::execution::{MatchOutcome, MatchResult, TakerKind}; use crate::orders::{Hash32, Id, OrderType, OrderUpdate, PegReferenceType, Side, TimeInForce}; use crate::price_level::PriceLevelSnapshotPackage; use crate::price_level::level::{PriceLevel, PriceLevelData}; @@ -7343,6 +7343,252 @@ mod tests { "from_snapshot must reject a mixed-side snapshot" ); } + + // ------------------------------------------------------------------ + // Issue #119 — atomic quantity-increase re-sequencing + // ------------------------------------------------------------------ + + #[test] + fn test_upsize_vs_cancel_race_no_lost_cancel_or_resurrection() { + use std::sync::{Arc as StdArc, Barrier}; + use std::thread; + + const ITERATIONS: usize = 1_500; + for iter in 0..ITERATIONS { + let level = StdArc::new(PriceLevel::new(10_000)); + level + .add_order(create_standard_order(1, 10_000, 100)) + .expect("seed maker"); + let id = Id::from_u64(1); + let barrier = StdArc::new(Barrier::new(2)); + + let updater = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + // Quantity INCREASE -> in-place demotion to the tail. + level + .update_order(OrderUpdate::UpdateQuantity { + order_id: id, + new_quantity: Quantity::new(200), + }) + .expect("update must not error") + }) + }; + let canceller = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + level + .update_order(OrderUpdate::Cancel { order_id: id }) + .expect("cancel must not error") + }) + }; + + let _ = updater.join().expect("updater panicked"); + let cancelled = canceller.join().expect("canceller panicked"); + + let is_resting = level + .snapshot_by_insertion_seq() + .iter() + .any(|o| o.id() == id); + + // The id never leaves the map, so the cancel is the only remover: it + // must always succeed (Some), and the order must then be gone. It is + // NEVER the case that the cancel returned None while the order still + // rests (the lost-cancel / resurrection bug the old remove+push + // demotion allowed). + assert!( + cancelled.is_some(), + "iter {iter}: cancel returned None (the order was momentarily absent — resurrection window)" + ); + assert!( + !is_resting, + "iter {iter}: order still resting after a winning cancel (resurrection)" + ); + assert_counters_match_queue(&level); + } + } + + #[test] + fn test_upsize_vs_duplicate_admission_race_always_rejected() { + use std::sync::{Arc as StdArc, Barrier}; + use std::thread; + + const ITERATIONS: usize = 1_500; + for iter in 0..ITERATIONS { + let level = StdArc::new(PriceLevel::new(10_000)); + level + .add_order(create_standard_order(1, 10_000, 100)) + .expect("seed maker"); + let id = Id::from_u64(1); + let barrier = StdArc::new(Barrier::new(2)); + + let updater = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + level + .update_order(OrderUpdate::UpdateQuantity { + order_id: id, + new_quantity: Quantity::new(200), + }) + .expect("update must not error") + }) + }; + let admitter = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + // Same id, distinct order. Since the upsize never vacates the + // id, this must ALWAYS be rejected as a duplicate. + level.add_order(create_standard_order(1, 10_000, 50)) + }) + }; + + let _ = updater.join().expect("updater panicked"); + let admit = admitter.join().expect("admitter panicked"); + + assert!( + matches!(admit, Err(PriceLevelError::DuplicateOrderId(_))), + "iter {iter}: duplicate admission must always be rejected (id never leaves the map); got {admit:?}" + ); + + // No counter drift, exactly one resting id, map/index 1:1. + assert_counters_match_queue(&level); + let ids: Vec = level + .snapshot_by_insertion_seq() + .iter() + .map(|o| o.id()) + .collect(); + assert_eq!(ids, vec![id], "iter {iter}: id 1 must rest exactly once"); + assert_eq!(level.order_count(), 1); + } + } + + #[test] + fn test_concurrent_upsize_and_match_stays_consistent() { + // Stress the stale-front-selection guard: one thread performs a bounded + // burst of upsizes (each a tail demotion that re-sequences the maker + // mid-sweep) while a single matcher races it with small takers, then + // fully drains once the burst ends. Invariants: never panics; every + // consumed maker id is a real maker (a stale front cannot act on a + // garbage / re-sequenced-away entry); and the level drains to empty with + // counters consistent. + use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering}; + use std::sync::{Arc as StdArc, Barrier, Mutex}; + use std::thread; + + const MAKERS: u64 = 6; + const PRICE: u128 = 10_000; + const UPSIZE_ROUNDS: usize = 150; + + for iter in 0..25 { + let level = StdArc::new(PriceLevel::new(PRICE)); + for id in 1..=MAKERS { + level + .add_order(create_standard_order(id, PRICE, 100)) + .expect("seed maker"); + } + let barrier = StdArc::new(Barrier::new(2)); + let burst_done = StdArc::new(AtomicBool::new(false)); + let consumed = StdArc::new(Mutex::new(Vec::::new())); + + let upsizer = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + let burst_done = StdArc::clone(&burst_done); + thread::spawn(move || { + barrier.wait(); + let mut q = 100u64; + // A BOUNDED burst so the level can actually be drained. + for _ in 0..UPSIZE_ROUNDS { + for id in 1..=MAKERS { + q += 1; + // Ignore the result: the maker may already be gone + // (consumed by the matcher), which returns Ok(None). + let _ = level.update_order(OrderUpdate::UpdateQuantity { + order_id: Id::from_u64(id), + new_quantity: Quantity::new(q), + }); + } + } + burst_done.store(true, AtomicOrdering::Release); + }) + }; + let matcher = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + let burst_done = StdArc::clone(&burst_done); + let consumed = StdArc::clone(&consumed); + thread::spawn(move || { + barrier.wait(); + let generator = UuidGenerator::new(Uuid::from_u128(0xBEEF_0000 + iter as u128)); + let record = |result: &MatchResult| { + let mut guard = consumed.lock().expect("lock"); + for trade in result.trades().as_vec() { + guard.push(trade.maker_order_id()); + } + }; + // Race the burst with small takers. + while !burst_done.load(AtomicOrdering::Acquire) { + let result = level.match_order( + 7, + Id::from_u64(9_999), + TimeInForce::Gtc, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ); + record(&result); + } + // Burst over: drain whatever remains with a large taker. + loop { + if level.order_count() == 0 { + break; + } + let result = level.match_order( + u64::MAX, + Id::from_u64(9_999), + TimeInForce::Gtc, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_001), + &generator, + ); + record(&result); + if result.trades().as_vec().is_empty() { + break; // safety: no progress + } + } + }) + }; + + upsizer.join().expect("upsizer panicked"); + matcher.join().expect("matcher panicked"); + + assert_eq!( + level.order_count(), + 0, + "iter {iter}: level must drain empty" + ); + assert_counters_match_queue(&level); + + // Every consumed maker id is one of the real makers — a stale front + // never surfaces a phantom / re-sequenced-away entry. + let expected: std::collections::HashSet = (1..=MAKERS).map(Id::from_u64).collect(); + let guard = consumed.lock().expect("lock"); + for maker in guard.iter() { + assert!( + expected.contains(maker), + "iter {iter}: consumed an unexpected maker id {maker}" + ); + } + } + } } #[cfg(test)] From e4cfe70cef7c57e98780ea6de4eeabf55487e2b6 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 17:20:34 +0200 Subject: [PATCH 4/4] address review: linearizable index re-key and sequence-validated pop - resequence_to_tail and the match sweep's ReplaceAtTail now publish the new index key BEFORE removing the old one, so a continuously resident maker is never transiently absent from the ordered index and a concurrent front scan can never report Empty past resting liquidity; the transient two-key window is discarded on selection by the stale-front guard (the stored sequence is already the new one) - pop_entry validates, under the map entry, that the stored sequence equals the popped index key; a demoted maker's stale old key is dropped and the scan retries, so the destructive pop can neither return a just-demoted maker ahead of older makers nor strand its new key - the allocation-free claim on the re-key path is narrowed: the SkipMap insert allocates the new index node Stress + deterministic tests: 200k front scans against a continuous demoter never observe Empty; a demoted maker drains last and exactly once under a racing pop; map/index end empty and consistent. --- src/price_level/order_queue.rs | 58 ++++++++--- src/price_level/tests/order_queue.rs | 149 +++++++++++++++++++++++++++ 2 files changed, 192 insertions(+), 15 deletions(-) diff --git a/src/price_level/order_queue.rs b/src/price_level/order_queue.rs index 72e5716..ba7a706 100644 --- a/src/price_level/order_queue.rs +++ b/src/price_level/order_queue.rs @@ -222,11 +222,26 @@ impl OrderQueue { loop { // `pop_front` atomically removes the lowest-sequence index entry. let entry = self.index.pop_front()?; + let popped_seq = *entry.key(); let order_id = *entry.value(); - // The id may have been concurrently cancelled via `remove`; in that - // case the map no longer holds it, so skip it and try the next one. - if let Some((_, (seq, order))) = self.orders.remove(&order_id) { - return Some((seq, order)); + // Validate the maker's STORED sequence against the key we popped, + // under the map entry lock (issue #127). A concurrent + // `resequence_to_tail` may have demoted this id to a fresh tail + // sequence, making this a STALE old key; removing by id alone would + // return the demoted maker ahead of older makers and strand its new + // key. Mirror `match_front`'s stale-front guard: only take the maker + // when the popped key IS its current key. + match self.orders.entry(order_id) { + Entry::Occupied(occupied) if occupied.get().0 == popped_seq => { + let (seq, order) = occupied.remove(); + return Some((seq, order)); + } + // Stale old key of a demoted maker (stored seq != popped), or the + // id was cancelled (`Vacant`). Either way the popped key is + // already gone from the index (`pop_front` removed it); the maker, + // if it still rests, lives under its newer key and is popped in + // order on a later iteration. Retry with the next front. + _ => continue, } } } @@ -396,7 +411,10 @@ impl OrderQueue { // `occupied` still holds the per-entry lock here, so // re-keying the index — a different structure // (`SkipMap`), no deadlock — happens while a concurrent - // cancel is still excluded from the entry. Once the + // cancel is still excluded from the entry. Insert the + // NEW key BEFORE removing the old (issue #127) so the + // id is never transiently absent from the index and a + // concurrent front scan can never miss it. Once the // lock is released the value already carries `new_seq`, // so a cancel removes `orders[id]` and `index[new_seq]` // consistently. The only residue a race can leave is a @@ -404,8 +422,8 @@ impl OrderQueue { // already-removed id, which the next `match_front` // self-heals on the `Vacant` branch. No order and no // counter update is ever lost. - self.index.remove(&seq); self.index.insert(new_seq, order_id); + self.index.remove(&seq); drop(occupied); } FrontAction::SetAside => { @@ -446,14 +464,15 @@ impl OrderQueue { /// [`OrderQueue::match_front`] could act on a stale front. Because the id /// never leaves the map here, a concurrent [`OrderQueue::remove`] either /// fully precedes this call (this returns `None`) or fully follows it (it - /// removes the re-sequenced order): all three hazards are closed. Note the - /// index re-key is still two SkipMap ops, so a concurrent front scan can - /// transiently miss the maker in the INDEX (map residency is unaffected) — - /// at worst a missed fill for that sweep, the same transient - /// [`FrontAction::ReplaceAtTail`] has always had and strictly narrower - /// than the old remove+push map absence. - /// - /// Allocation-free beyond the `Arc` the caller hands in. + /// removes the re-sequenced order): all three hazards are closed. The index + /// re-key inserts the NEW key before removing the old (issue #127), so the id + /// is never absent from the INDEX either: a concurrent front scan always + /// finds this maker at one key or the other and can never return `Empty` + /// with liquidity resting. The transient two-key window is discarded by the + /// stale-front guard in [`OrderQueue::match_front`] / [`OrderQueue::pop_entry`]. + /// + /// Allocation-free beyond the `Arc` the caller hands in and the index node + /// [`crossbeam_skiplist::SkipMap::insert`] allocates for the new key. pub(crate) fn resequence_to_tail( &self, order_id: Id, @@ -475,8 +494,17 @@ impl OrderQueue { slot.0 = new_seq; (old_seq, replaced) }; - self.index.remove(&old_seq); + // Re-key the index NEW-KEY-FIRST (issue #127): insert the new + // sequence BEFORE removing the old one, so the id is never + // transiently absent from the index. A concurrent `match_front` + // front scan therefore always finds this maker (at the old key or + // the new one) — it can never see a gap and return `Empty` with + // liquidity resting. The transient window where BOTH keys point + // at the id is harmless: the stale-front guard (`stored != seq`) + // in `match_front` / `pop_entry` discards the old key on + // selection, since the stored sequence is already `new_seq`. self.index.insert(new_seq, order_id); + self.index.remove(&old_seq); Some(replaced) } } diff --git a/src/price_level/tests/order_queue.rs b/src/price_level/tests/order_queue.rs index 061ab65..705882c 100644 --- a/src/price_level/tests/order_queue.rs +++ b/src/price_level/tests/order_queue.rs @@ -610,4 +610,153 @@ mod tests { assert!(queue.is_empty(), "iter {iter}: queue must fully drain"); } } + + #[test] + fn test_resequence_vs_front_scan_never_empty() { + // Issue #127 🔴: the index re-key inserts the new key BEFORE removing the + // old, so a resident maker is never transiently absent from the index. A + // front scan (`match_front`) racing continuous demotions of that maker + // must therefore NEVER return `Empty` — the liquidity is always resting. + // Under the old remove-then-insert order the maker vanished from the + // index between the two ops and a scan could miss it. + use crate::price_level::order_queue::{FrontAction, FrontOutcome}; + use std::collections::HashSet; + use std::sync::Arc as StdArc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::thread; + + let queue = StdArc::new(OrderQueue::new()); + // The single resident maker (id 1) never leaves the queue. + queue.push(StdArc::new(create_test_order(1, 1_000, 100))); + let done = StdArc::new(AtomicBool::new(false)); + + let demoter = { + let queue = StdArc::clone(&queue); + let done = StdArc::clone(&done); + thread::spawn(move || { + while !done.load(Ordering::Relaxed) { + // Demote the resident maker to a fresh tail sequence, over and + // over. It stays resident the whole time. + let _ = queue.resequence_to_tail( + Id::from_u64(1), + StdArc::new(create_test_order(1, 1_000, 100)), + ); + } + }) + }; + + for _ in 0..200_000 { + let mut set_aside = HashSet::new(); + // A no-op probe: whatever the front is, park it (leaves it resting) + // and report we found one. The maker always rests, so this must be + // `Matched`, never `Empty`. + let outcome = + queue.match_front(&mut set_aside, |_seq, _order| (FrontAction::SetAside, ())); + assert!( + matches!(outcome, FrontOutcome::Matched { .. }), + "front scan returned Empty while the resident maker rests (issue #127)" + ); + } + + done.store(true, Ordering::Relaxed); + demoter.join().expect("demoter thread panicked"); + assert_eq!(queue.len(), 1, "the resident maker still rests"); + assert!( + queue.debug_map_index_consistent(), + "map and index must be 1:1 after the demotions" + ); + } + + #[test] + fn test_pop_entry_after_resequence_orders_last_and_drains_clean() { + // Issue #127 🔴: after a maker is demoted, `pop_entry` must return it + // LAST (at its new tail sequence), never early via a stale old key, and a + // full drain must leave BOTH the map and the ordered index empty. + let queue = OrderQueue::new(); + queue.push(Arc::new(create_test_order(1, 1_000, 10))); // seq 0 + queue.push(Arc::new(create_test_order(2, 1_000, 20))); // seq 1 (to demote) + queue.push(Arc::new(create_test_order(3, 1_000, 30))); // seq 2 + + let replaced = + queue.resequence_to_tail(Id::from_u64(2), Arc::new(create_test_order(2, 1_000, 25))); + assert!(replaced.is_some(), "the demoted maker was resident"); + // New-before-old re-key leaves no stale old key: map and index are 1:1. + assert!(queue.debug_map_index_consistent()); + + let mut ids = Vec::new(); + while let Some((_, order)) = queue.pop_entry() { + ids.push(order.id()); + } + // FIFO by CURRENT sequence: 1, 3, then the demoted 2 LAST. + assert_eq!( + ids, + vec![Id::from_u64(1), Id::from_u64(3), Id::from_u64(2)], + "the demoted maker must pop last, never via its stale old key" + ); + // Fully drained: map and index both empty (and consistent). + assert!(queue.is_empty()); + assert!( + queue.debug_map_index_consistent(), + "no stale index entry may survive a full drain" + ); + } + + #[test] + fn test_pop_entry_vs_resequence_race_drains_each_once() { + // Issue #127 🔴: `pop_entry` validates the stored sequence against the + // popped key, so under continuous demotions racing a drain, every id + // comes out EXACTLY once (never twice via a stale old key, never lost), + // and the queue drains clean. + use std::sync::Arc as StdArc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::thread; + + const N: u64 = 50; + + for iter in 0..400 { + let queue = StdArc::new(OrderQueue::new()); + for id in 1..=N { + queue.push(StdArc::new(create_test_order(id, 1_000, 10))); + } + let done = StdArc::new(AtomicBool::new(false)); + let demoter = { + let queue = StdArc::clone(&queue); + let done = StdArc::clone(&done); + thread::spawn(move || { + while !done.load(Ordering::Relaxed) { + // Continuously demote a mid maker; once it is popped this + // returns `None` (id gone) and simply spins. + let _ = queue.resequence_to_tail( + Id::from_u64(N / 2), + StdArc::new(create_test_order(N / 2, 1_000, 10)), + ); + } + }) + }; + + let mut popped: Vec = Vec::new(); + while let Some((_, order)) = queue.pop_entry() { + popped.push(order.id()); + } + + done.store(true, Ordering::Relaxed); + demoter.join().expect("demoter thread panicked"); + + let unique: std::collections::HashSet = popped.iter().copied().collect(); + assert_eq!( + unique.len(), + popped.len(), + "iter {iter}: every id popped at most once (no stale-key double pop)" + ); + assert_eq!( + popped.len() as u64, + N, + "iter {iter}: all ids drained exactly once (none lost)" + ); + assert!( + queue.is_empty() && queue.debug_map_index_consistent(), + "iter {iter}: clean drain, map and index both empty" + ); + } + } }