From 9edab14ea8997032d058064aaf1f7881e55537a7 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 12:57:46 +0200 Subject: [PATCH 1/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 2/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" + ); + } + } } From b097f8b56324441ead433df21bde5da73880803b Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 13:26:51 +0200 Subject: [PATCH 3/4] fix(price_level): apply UpdateQuantity to the live maker state UpdateQuantity derived the replacement and its priority policy from a stale pre-read: a concurrent match could consume or replenish the maker between the read and the swap, so the stale replacement resurrected already-executed visible or hidden quantity (iceberg hidden depth included), and the stale total could pick the wrong priority branch. Counters stayed numerically consistent while describing resurrected liquidity, and a snapshot persisted the corruption durably. New OrderQueue::update_entry runs the decide closure against the LIVE stored order under the per-entry lock and commits its UpdateDecision (in-place swap, or inline tail re-sequence -- a separate call would double-lock the shard) in the same critical section. The level closure resizes from the live order (new_quantity applies to the live visible tranche, hidden preserved as live -- resurrection is structurally impossible), decides increase-vs-decrease from the live total, and reserves the level-counter delta with fetch_update(checked_add) BEFORE the queue commits: an update whose delta would overflow a level counter is rejected with InvalidOperation leaving maker, position, and counters untouched. That closes the fetch_add(new - old) wrap flagged in the issue-111 audit. update_in_place and resequence_to_tail are subsumed and removed. Race tests (Barrier, 1000 iterations each): competing same-id updates leave exactly one winner with consistent counters; decrease-vs-cancel never resurrects; update-vs-match on a replenishing iceberg never raises hidden above its pre-race value; plus a deterministic level-counter overflow rejection and a live-derivation sanity case. The shared consistency helper now also asserts the LIVE advisory counters converge to the queue sums when quiescent (previously it only compared snapshot folds, which are consistent by construction). Perf: update-heavy bench 24.12us vs 24.97us median -- ~3% faster (one entry-lock acquisition instead of a pre-read plus a separate commit). Closes #115 --- src/price_level/level.rs | 211 ++++++++--------- src/price_level/order_queue.rs | 203 +++++++++-------- src/price_level/tests/level.rs | 327 +++++++++++++++++++++++++++ src/price_level/tests/order_queue.rs | 35 ++- 4 files changed, 552 insertions(+), 224 deletions(-) diff --git a/src/price_level/level.rs b/src/price_level/level.rs index 7a0645c..2824e1d 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -4,7 +4,7 @@ use crate::UuidGenerator; use crate::errors::PriceLevelError; use crate::execution::{MatchResult, TakerKind, Trade}; use crate::orders::{Id, OrderType, OrderUpdate, Side, TimeInForce}; -use crate::price_level::order_queue::{FrontAction, FrontOutcome, OrderQueue}; +use crate::price_level::order_queue::{FrontAction, FrontOutcome, OrderQueue, UpdateDecision}; use crate::price_level::{PriceLevelSnapshot, PriceLevelSnapshotPackage, PriceLevelStatistics}; use crate::utils::{Price, Quantity, TimestampMs}; use serde::{Deserialize, Serialize}; @@ -1532,14 +1532,27 @@ impl PriceLevel { /// visible tranche and keep hidden), so the branch reflects the real size /// change rather than a silent no-op. /// + /// # Applied to the live maker (issue #115) + /// + /// The resize, the priority decision, and the level-counter update are all + /// derived from the order **currently resident in the queue**, computed and + /// committed together under a single per-entry lock — never from a pre-read. + /// A concurrent match or replenishment that commits first is therefore fully + /// reflected: an update applies `new_quantity` to the *live* visible tranche + /// and preserves the *live* hidden depth, so it can never resurrect executed + /// or cancelled quantity, and the increase-vs-decrease policy is chosen from + /// the live total (never stale). The level counters are validated with + /// checked math before the queue mutates, so an update that would overflow a + /// level counter is rejected with the level left unchanged. + /// /// # Errors /// /// Returns [`PriceLevelError::InvalidOperation`] if an /// [`OrderUpdate::UpdatePrice`] / [`OrderUpdate::Replace`] would not move /// the order to a different price level, if computing an order's total - /// quantity overflows `u64`, or if applying a quantity update's counter - /// delta would take the level's visible- or hidden-quantity counter past - /// `u64::MAX` (rejected with the queue and counters untouched). + /// quantity overflows `u64`, or if an [`OrderUpdate::UpdateQuantity`] would + /// overflow the level's visible- or hidden-quantity counter (the maker and + /// its queue position are left unchanged in that case). #[must_use = "the updated order (or None when the order is absent) must be handled"] pub fn update_order( &self, @@ -1591,126 +1604,96 @@ impl PriceLevel { order_id, new_quantity, } => { - // Read the current order to build the resized order and pick the - // priority policy. The counter deltas below are taken from the - // order actually removed/replaced under the queue's per-entry - // lock — not from this pre-read — so a concurrent update cannot - // drift `visible_quantity` / `hidden_quantity` from the queue. - let Some(order) = self.orders.find(order_id) else { - return Ok(None); // Order not found - }; - - let prev_total = order - .visible_quantity() - .as_u64() - .checked_add(order.hidden_quantity().as_u64()) - .ok_or_else(|| PriceLevelError::InvalidOperation { - message: "order total quantity overflow".to_string(), - })?; - - // Build the updated order. `with_reduced_quantity` sets the - // (visible/main) quantity to exactly `new_quantity` for every - // order variant, so `new_total` reflects the real post-update - // size and the increase/decrease branch below is chosen - // correctly for all types. - let new_order = order.with_reduced_quantity(new_quantity.as_u64()); - let new_visible = new_order.visible_quantity().as_u64(); - let new_hidden = new_order.hidden_quantity().as_u64(); - let new_total = new_visible.checked_add(new_hidden).ok_or_else(|| { - PriceLevelError::InvalidOperation { - message: "order total quantity overflow".to_string(), + // Overflow-checked forward reservation of a level counter for a + // component moving `old -> new`. Increase is validated with + // `checked_add` (rejecting before any queue mutation); decrease + // simply subtracts. `Relaxed`: advisory counters (issue #68). + fn reserve( + counter: &std::sync::atomic::AtomicU64, + old: u64, + new: u64, + ) -> Result<(), PriceLevelError> { + if new >= old { + counter + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + c.checked_add(new - old) + }) + .map(|_| ()) + .map_err(|_| PriceLevelError::InvalidOperation { + message: "price level quantity counter overflow on update" + .to_string(), + }) + } else { + counter.fetch_sub(old - new, Ordering::Relaxed); + Ok(()) } - })?; - - // Pre-validate the LEVEL counter deltas before mutating the - // queue: an upsize (or a visible/hidden reshuffle) whose delta - // would take a level counter past `u64::MAX` must be rejected - // with the queue and counters untouched, never committed and then - // wrapped by the raw `fetch_add` below. Project each counter from - // its live value using the pre-read order's components as `old`; - // if either projection does not fit `u64`, reject. - // - // This uses the pre-read `order` (not the value the queue mutation - // will actually replace), so a concurrent update of the same id - // could make the projection stale (TOCTOU) — strictly better than - // the unchecked wrap it replaces, and #115 closes the window - // exactly by reserving under the entry lock. - let old_visible_pre = order.visible_quantity().as_u64(); - let old_hidden_pre = order.hidden_quantity().as_u64(); - let projects = |counter: &AtomicU64, old: u64, new: u64| -> bool { - let cur = counter.load(Ordering::Relaxed); + } + // Undo this call's own `old -> new` reservation (commutative with + // concurrent deltas — it reverses exactly what it added). + fn unreserve(counter: &std::sync::atomic::AtomicU64, old: u64, new: u64) { if new >= old { - cur.checked_add(new - old).is_some() + counter.fetch_sub(new - old, Ordering::Relaxed); } else { - cur.checked_sub(old - new).is_some() + counter.fetch_add(old - new, Ordering::Relaxed); } - }; - if !projects(&self.visible_quantity, old_visible_pre, new_visible) - || !projects(&self.hidden_quantity, old_hidden_pre, new_hidden) - { - return Err(PriceLevelError::InvalidOperation { - message: "price level quantity counter overflow on update".to_string(), - }); } - let new_order_arc = Arc::new(new_order); - - // Perform the queue mutation and capture the order it actually - // 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 (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. - }; - replaced - } else { - // Quantity DECREASE or unchanged total: keep the maker's - // queue position by swapping the stored order in place at its - // existing insertion sequence, under the DashMap per-entry - // lock. - let Some(replaced) = - self.orders.update_in_place(order_id, new_order_arc.clone()) - else { - return Ok(None); // Removed by another thread. - }; - replaced - }; + let visible_counter = &self.visible_quantity; + let hidden_counter = &self.hidden_quantity; + + // Derive the resized order, choose the priority policy, and + // reserve the level counters ALL against the LIVE stored order, + // under the entry lock (issue #115). Nothing is read before the + // lock, so a concurrent match / replenish that committed first is + // fully reflected: the update can never resurrect executed or + // cancelled visible / hidden quantity, and the policy is chosen + // from the live total, not a stale pre-read. The counters are + // reserved with checked math BEFORE the queue commits, so an + // update that would overflow a level counter is rejected with the + // level (and queue) untouched. + let outcome = self.orders.update_entry(order_id, |live| { + let old_visible = live.visible_quantity().as_u64(); + let old_hidden = live.hidden_quantity().as_u64(); + let live_total = old_visible.checked_add(old_hidden).ok_or_else(|| { + PriceLevelError::InvalidOperation { + message: "order total quantity overflow".to_string(), + } + })?; - // Apply the counter deltas from the actual replaced order. A - // single component (visible or hidden) can move in EITHER - // direction even when the total shrinks or is unchanged, because - // quantity can shift between the visible and hidden portions, so - // handle both signs. - 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` / `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 { - counter.fetch_add(new - old, Ordering::Relaxed); + // `with_reduced_quantity` sets the visible/main tranche to + // exactly `new_quantity` for every variant and preserves the + // LIVE hidden depth (never restored from a pre-read). + let new_order = live.with_reduced_quantity(new_quantity.as_u64()); + let new_visible = new_order.visible_quantity().as_u64(); + let new_hidden = new_order.hidden_quantity().as_u64(); + let new_total = new_visible.checked_add(new_hidden).ok_or_else(|| { + PriceLevelError::InvalidOperation { + message: "order total quantity overflow".to_string(), + } + })?; + + // Validate + reserve the level counters before mutating the + // queue. On a hidden overflow, roll the visible reservation + // back so a rejected update leaves the counters unchanged. + reserve(visible_counter, old_visible, new_visible)?; + if let Err(err) = reserve(hidden_counter, old_hidden, new_hidden) { + unreserve(visible_counter, old_visible, new_visible); + return Err(err); + } + + // Priority policy from the LIVE total (cannot be stale). + let arc = Arc::new(new_order); + if new_total > live_total { + Ok(UpdateDecision::ReplaceAtTail(arc)) } else { - counter.fetch_sub(old - new, Ordering::Relaxed); + Ok(UpdateDecision::KeepInPlace(arc)) } - }; - apply(&self.visible_quantity, old_visible, new_visible); - apply(&self.hidden_quantity, old_hidden, new_hidden); + }); - Ok(Some(new_order_arc)) + match outcome { + None => Ok(None), // Order not found / concurrently removed. + Some(result) => result.map(Some), + } } OrderUpdate::UpdatePriceAndQuantity { diff --git a/src/price_level/order_queue.rs b/src/price_level/order_queue.rs index ba7a706..ef94799 100644 --- a/src/price_level/order_queue.rs +++ b/src/price_level/order_queue.rs @@ -62,6 +62,22 @@ pub(crate) enum FrontAction { SetAside, } +/// The mutation an [`OrderQueue::update_entry`] decision closure asks the queue +/// to commit, after deriving it from the **live** stored order under the entry +/// lock. Mirrors the [`FrontAction`] precedent for the match sweep. +#[derive(Debug)] +pub(crate) enum UpdateDecision { + /// Decrease / unchanged total: swap the stored value to the resized order at + /// its existing insertion sequence, keeping its price-time position. + KeepInPlace(Arc>), + /// Increase in total: demote the resized order to a fresh tail sequence + /// (losing time priority) by minting a new sequence, swapping the stored + /// `(seq, order)` pair in place, and re-keying the index — all under the + /// entry lock the update already holds. Same shape as the + /// [`FrontAction::ReplaceAtTail`] the match sweep commits. + ReplaceAtTail(Arc>), +} + /// The outcome of a single [`OrderQueue::match_front`] step, reported back to /// the sweep so it can drive the loop and apply counter deltas. #[derive(Debug)] @@ -93,11 +109,12 @@ impl OrderQueue { /// /// 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)]`. + /// (insert-if-absent, issue #113) and every quantity update re-derives and + /// re-sequences in place under the entry lock via + /// [`OrderQueue::update_entry`] (issue #115). 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 @@ -119,9 +136,10 @@ 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. + /// The quantity-update path no longer vacates the id either: as of issues + /// #119 / #115 it re-derives and re-sequences in place under the entry lock + /// (`update_entry`), so there is no remove-then-push window for a same-id + /// admission to slip into. /// /// # Errors /// @@ -340,9 +358,10 @@ impl OrderQueue { 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 + // taken. A concurrent quantity-increase demotion (the + // `ReplaceAtTail` path of `update_entry`) 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 @@ -446,66 +465,83 @@ 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. 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( + /// Atomically derive, decide, and commit an update against the **live** + /// stored order for `order_id`, all inside the per-entry lock (issue #115). + /// + /// `decide` runs against the order currently resident in the map — not a + /// stale pre-read — and returns the [`UpdateDecision`] to commit, or a + /// [`PriceLevelError`] to reject the update without touching the queue. The + /// decision (the resized order and the in-place-vs-demote priority policy) + /// therefore reflects any concurrent match / replenish that committed before + /// the lock was taken, so an update can never resurrect executed or + /// cancelled quantity, and the priority policy can never be chosen from a + /// stale total. This mirrors the [`OrderQueue::match_front`] decision-closure + /// pattern; the closure must not let a reference into the live order escape + /// its return value. + /// + /// Returns: + /// - `None` if the id is not present (concurrently removed / never existed); + /// - `Some(Err(_))` if `decide` rejected the update (queue untouched); + /// - `Some(Ok(new_order))` with the committed order on success. + /// + /// Both commits happen under the single entry lock this method already + /// holds: `KeepInPlace` swaps the stored value; `ReplaceAtTail` mints a fresh + /// tail sequence, swaps the `(seq, order)` pair, and re-keys the index in + /// place (delegating to a separate re-sequence method here would deadlock on + /// the same shard lock). + #[must_use = "the caller must handle committed / rejected / absent outcomes"] + pub(crate) fn update_entry( &self, order_id: Id, - new_order: Arc>, - ) -> Option>> { + decide: F, + ) -> Option>, PriceLevelError>> + where + F: FnOnce(&OrderType<()>) -> Result, + { 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) + // Derive + decide against the LIVE stored order under the lock. + // The borrow ends with the `decide` call (it returns owned data), + // so `get_mut()` below is free to commit. + let decision = match decide(occupied.get().1.as_ref()) { + Ok(decision) => decision, + Err(err) => return Some(Err(err)), + }; + debug_assert_eq!( + match &decision { + UpdateDecision::KeepInPlace(o) | UpdateDecision::ReplaceAtTail(o) => o.id(), + }, + order_id, + "update_entry: the decided order must keep the id it is stored under" + ); + let committed = match decision { + UpdateDecision::KeepInPlace(new_order) => { + occupied.get_mut().1 = new_order.clone(); + new_order + } + UpdateDecision::ReplaceAtTail(new_order) => { + let new_seq = self.next_seq.fetch_add(1, Ordering::Relaxed); + let old_seq = { + let slot = occupied.get_mut(); + let old_seq = slot.0; + slot.0 = new_seq; + slot.1 = new_order.clone(); + old_seq + }; + // Re-key NEW-KEY-FIRST (issue #127): insert the new + // sequence before removing the old one, so the id is + // never transiently absent from the index and a + // concurrent front scan can never return `Empty` with + // liquidity resting. The transient two-key window is + // discarded on selection by the stale-front guard + // (the stored sequence is already `new_seq`). + self.index.insert(new_seq, order_id); + self.index.remove(&old_seq); + new_order + } }; - // 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) + Some(Ok(committed)) } } } @@ -534,35 +570,6 @@ impl OrderQueue { self.orders.get(&order_id).map(|o| o.value().1.clone()) } - /// Replace the stored order for `order_id` in place, keeping its existing - /// insertion sequence (and therefore its price-time / FIFO position). - /// - /// Returns the previous order if `order_id` was present, or `None` if it - /// was not (e.g. concurrently removed). The `index` entry `seq -> id` - /// stays valid because the sequence is unchanged, so only the `DashMap` - /// value is swapped. - /// - /// The whole swap happens under the `DashMap` per-entry lock, so a - /// concurrent [`OrderQueue::remove`] of the same id either observes the - /// old value and removes it, or observes the new value and removes that — - /// it never sees the entry mid-update. This closes the - /// absent-from-`orders` window that a remove-then-push sequence would open. - #[must_use] - pub(crate) fn update_in_place( - &self, - order_id: Id, - new_order: Arc>, - ) -> Option>> { - debug_assert_eq!( - new_order.id(), - order_id, - "update_in_place: new_order id must match the key it is stored under" - ); - let mut entry = self.orders.get_mut(&order_id)?; - let (_seq, slot) = entry.value_mut(); - Some(std::mem::replace(slot, new_order)) - } - /// Remove an order with the given ID. /// Returns the removed order if found. Cleans both the map and the index. #[must_use] @@ -633,8 +640,8 @@ impl OrderQueue { /// stored sequence. Because the map holds exactly one entry per order, a /// duplicate id is impossible by construction, and because each /// `(seq, order)` pair is swapped atomically under the `DashMap` per-entry - /// lock (both [`FrontAction::ReplaceAtTail`] and - /// [`OrderQueue::update_in_place`] mutate value and sequence together under + /// lock (both the [`FrontAction::ReplaceAtTail`] sweep step and + /// [`OrderQueue::update_entry`] mutate value and sequence together under /// it), every emitted pair is a real committed state — an order caught /// mid-re-sequencing appears at either its old or its new sequence, never /// both and never as a mixed `(old_seq, new_order)` pair. Walking the @@ -678,8 +685,8 @@ impl OrderQueue { .map(|entry| entry.value().clone()) .collect(); // Unstable sort is deterministic here because sequences are unique - // across live orders (`push` / `ReplaceAtTail` mint distinct seqs via - // `fetch_add`; `update_in_place` keeps the order's own seq). + // across live orders (the tail-appending paths mint distinct seqs via + // `fetch_add`; an in-place update keeps the order's own seq). pairs.sort_unstable_by_key(|(seq, _)| *seq); out.clear(); out.extend(pairs.into_iter().map(|(_, order)| order)); diff --git a/src/price_level/tests/level.rs b/src/price_level/tests/level.rs index 256c135..c96a554 100644 --- a/src/price_level/tests/level.rs +++ b/src/price_level/tests/level.rs @@ -5352,6 +5352,27 @@ mod tests { orders.len(), "order_count must equal the snapshot's own order-list length" ); + + // In the quiescent state (no concurrent writers — every race test joins + // its threads before asserting) the LIVE advisory atomics must have + // converged to the queue sums too. The snapshot-only assertions above + // are fold==fold by construction; these are the ones that catch a + // leaked or dropped counter reservation (issue #115). + assert_eq!( + level.visible_quantity(), + visible_sum, + "live visible counter must converge to the queue sum when quiescent" + ); + assert_eq!( + level.hidden_quantity(), + hidden_sum, + "live hidden counter must converge to the queue sum when quiescent" + ); + assert_eq!( + level.order_count(), + orders.len(), + "live order_count must converge to the queue length when quiescent" + ); } #[test] @@ -7589,6 +7610,312 @@ mod tests { } } } + + // ------------------------------------------------------------------ + // Issue #115 — UpdateQuantity applied to the live maker state + // ------------------------------------------------------------------ + + #[test] + fn test_update_quantity_level_counter_overflow_rejected() { + // Two Buy makers push the level's visible counter to just below u64::MAX; + // an upsize whose delta would carry it over must be rejected, with the + // maker, its queue position, and the counters all unchanged. + let level = PriceLevel::new(10_000); + level + .add_order(create_standard_order(1, 10_000, u64::MAX - 100)) + .expect("seed maker 1"); + level + .add_order(create_standard_order(2, 10_000, 50)) + .expect("seed maker 2"); + // Level visible counter == u64::MAX - 50. + + let before_json = level.snapshot_to_json().expect("snapshot before"); + let before_ids: Vec = level + .snapshot_by_insertion_seq() + .iter() + .map(|o| o.id()) + .collect(); + + // Upsize maker 2: 50 -> 200 (delta +150) would overflow the counter. + match level.update_order(OrderUpdate::UpdateQuantity { + order_id: Id::from_u64(2), + new_quantity: Quantity::new(200), + }) { + Err(PriceLevelError::InvalidOperation { message }) => { + assert!( + message.contains("overflow"), + "unexpected message: {message}" + ); + } + other => panic!("expected level-counter-overflow InvalidOperation, got {other:?}"), + } + + // Level byte-identical; maker 2 unchanged and in the same position. + assert_eq!( + level.snapshot_to_json().expect("snapshot after"), + before_json, + "a rejected update must leave the level unchanged" + ); + let after_ids: Vec = level + .snapshot_by_insertion_seq() + .iter() + .map(|o| o.id()) + .collect(); + assert_eq!(after_ids, before_ids); + let m2 = level + .snapshot_by_insertion_seq() + .into_iter() + .find(|o| o.id() == Id::from_u64(2)) + .expect("maker 2 still rests"); + assert_eq!(m2.visible_quantity().as_u64(), 50); + } + + #[test] + fn test_update_quantity_derives_from_live_iceberg_after_partial_fill() { + // Sequential sanity that an update resizes the LIVE visible tranche and + // preserves the LIVE hidden depth (the contract the concurrent path + // upholds): after a partial fill + replenish, the update must not restore + // the pre-fill hidden. + let level = PriceLevel::new(10_000); + level + .add_order(create_buy_iceberg_order(1, 10_000, 50, 100)) + .expect("seed iceberg"); + + let namespace = Uuid::parse_str("6ba7b810-9dad-11d1-80b4-00c04fd430c8").unwrap(); + let generator = UuidGenerator::new(namespace); + // Consume the full visible tranche (50): replenishes 50 from hidden, so + // the live maker becomes visible 50, hidden 50. + let _ = level.match_order( + 50, + Id::from_u64(999), + TimeInForce::Gtc, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ); + let live = level + .snapshot_by_insertion_seq() + .into_iter() + .find(|o| o.id() == Id::from_u64(1)) + .expect("iceberg rests"); + assert_eq!( + live.hidden_quantity().as_u64(), + 50, + "precondition: hidden drawn to 50" + ); + + // Resize visible to 30. Hidden must stay the LIVE 50, not the pre-fill 100. + level + .update_order(OrderUpdate::UpdateQuantity { + order_id: Id::from_u64(1), + new_quantity: Quantity::new(30), + }) + .expect("update ok") + .expect("maker present"); + let updated = level + .snapshot_by_insertion_seq() + .into_iter() + .find(|o| o.id() == Id::from_u64(1)) + .expect("iceberg rests"); + assert_eq!(updated.visible_quantity().as_u64(), 30); + assert_eq!( + updated.hidden_quantity().as_u64(), + 50, + "hidden must reflect the live 50, never resurrect the pre-fill 100" + ); + assert_counters_match_queue(&level); + } + + #[test] + fn test_competing_updates_same_id_one_winner() { + use std::sync::{Arc as StdArc, Barrier}; + use std::thread; + + const ITERATIONS: usize = 1_000; + 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 a = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + let _ = level.update_order(OrderUpdate::UpdateQuantity { + order_id: id, + new_quantity: Quantity::new(200), + }); + }) + }; + let b = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + thread::spawn(move || { + barrier.wait(); + let _ = level.update_order(OrderUpdate::UpdateQuantity { + order_id: id, + new_quantity: Quantity::new(50), + }); + }) + }; + a.join().expect("a panicked"); + b.join().expect("b panicked"); + + // Exactly one order rests, with one of the two requested quantities, + // and the counters agree with the queue. + let resting = level.snapshot_by_insertion_seq(); + assert_eq!(resting.len(), 1, "iter {iter}: exactly one maker rests"); + assert_eq!(resting[0].id(), id); + let q = resting[0].visible_quantity().as_u64(); + assert!( + q == 200 || q == 50, + "iter {iter}: final quantity {q} not one of the two updates" + ); + assert_counters_match_queue(&level); + } + } + + #[test] + fn test_update_decrease_vs_cancel_no_resurrection() { + use std::sync::{Arc as StdArc, Barrier}; + use std::thread; + + const ITERATIONS: usize = 1_000; + 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(); + // DECREASE branch (keeps sequence, in-place swap). + level.update_order(OrderUpdate::UpdateQuantity { + order_id: id, + new_quantity: Quantity::new(40), + }) + }) + }; + 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 cancel is the only remover; it always wins and the order is + // gone — never cancel-None-yet-resting (resurrection). + assert!( + cancelled.is_some(), + "iter {iter}: cancel returned None (resurrection window)" + ); + assert!( + !is_resting, + "iter {iter}: order still resting after a winning cancel" + ); + assert_counters_match_queue(&level); + } + } + + #[test] + fn test_update_vs_match_iceberg_stays_consistent() { + // An iceberg maker races a matcher (drawing visible, replenishing from + // hidden) against an updater (resizing visible). Invariants: never + // panics; hidden never exceeds its pre-race value (an update preserves + // live hidden, a match only draws it down — it is never resurrected); + // and the level stays consistent (counters == queue == snapshot). + use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering}; + use std::sync::{Arc as StdArc, Barrier}; + use std::thread; + + const INITIAL_HIDDEN: u64 = 400; + const UPDATE_ROUNDS: usize = 200; + + for iter in 0..25 { + let level = StdArc::new(PriceLevel::new(10_000)); + level + .add_order(create_buy_iceberg_order(1, 10_000, 20, INITIAL_HIDDEN)) + .expect("seed iceberg"); + let id = Id::from_u64(1); + let barrier = StdArc::new(Barrier::new(2)); + let done = StdArc::new(AtomicBool::new(false)); + + let updater = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + let done = StdArc::clone(&done); + thread::spawn(move || { + barrier.wait(); + for r in 0..UPDATE_ROUNDS { + let new_visible = 5 + (r as u64 % 30); + let _ = level.update_order(OrderUpdate::UpdateQuantity { + order_id: id, + new_quantity: Quantity::new(new_visible), + }); + } + done.store(true, AtomicOrdering::Release); + }) + }; + let matcher = { + let level = StdArc::clone(&level); + let barrier = StdArc::clone(&barrier); + let done = StdArc::clone(&done); + thread::spawn(move || { + barrier.wait(); + let generator = UuidGenerator::new(Uuid::from_u128(0xF00D_0000 + iter as u128)); + let mut max_hidden = 0u64; + while !done.load(AtomicOrdering::Acquire) { + let _ = level.match_order( + 3, + Id::from_u64(999), + TimeInForce::Gtc, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ); + // Sample the resting maker's hidden depth (if any). + if let Some(o) = level + .snapshot_by_insertion_seq() + .into_iter() + .find(|o| o.id() == id) + { + max_hidden = max_hidden.max(o.hidden_quantity().as_u64()); + } + } + max_hidden + }) + }; + + updater.join().expect("updater panicked"); + let max_hidden = matcher.join().expect("matcher panicked"); + + assert!( + max_hidden <= INITIAL_HIDDEN, + "iter {iter}: hidden {max_hidden} exceeded its pre-race value {INITIAL_HIDDEN} (resurrection)" + ); + assert_counters_match_queue(&level); + } + } } #[cfg(test)] diff --git a/src/price_level/tests/order_queue.rs b/src/price_level/tests/order_queue.rs index 705882c..22818f9 100644 --- a/src/price_level/tests/order_queue.rs +++ b/src/price_level/tests/order_queue.rs @@ -619,7 +619,7 @@ mod tests { // 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 crate::price_level::order_queue::{FrontAction, FrontOutcome, UpdateDecision}; use std::collections::HashSet; use std::sync::Arc as StdArc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -637,10 +637,11 @@ mod tests { 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)), - ); + let _ = queue.update_entry(Id::from_u64(1), |_live| { + Ok(UpdateDecision::ReplaceAtTail(StdArc::new( + create_test_order(1, 1_000, 100), + ))) + }); } }) }; @@ -672,14 +673,22 @@ mod tests { // 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. + use crate::price_level::order_queue::UpdateDecision; + 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"); + let replaced = queue.update_entry(Id::from_u64(2), |_live| { + Ok(UpdateDecision::ReplaceAtTail(Arc::new(create_test_order( + 2, 1_000, 25, + )))) + }); + assert!( + matches!(replaced, Some(Ok(_))), + "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()); @@ -703,6 +712,7 @@ mod tests { #[test] fn test_pop_entry_vs_resequence_race_drains_each_once() { + use crate::price_level::order_queue::UpdateDecision; // 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), @@ -726,10 +736,11 @@ mod tests { 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 _ = queue.update_entry(Id::from_u64(N / 2), |_live| { + Ok(UpdateDecision::ReplaceAtTail(StdArc::new( + create_test_order(N / 2, 1_000, 10), + ))) + }); } }) }; From a97a3f98230d20e76e7cc53da8e3aa505dd78ec3 Mon Sep 17 00:00:00 2001 From: Joaquin Bejar Date: Tue, 14 Jul 2026 17:58:05 +0200 Subject: [PATCH 4/4] address review: entry-locked replenish counters, deferred Arc drops - the match sweep's replenish branch now publishes its level-counter transition inside the locked decision closure -- one checked fetch_update applying the net visible delta plus the hidden draw -- so the counters can never transiently lag a committed replenishment and a concurrent UpdateQuantity decrease can no longer underflow; the post-lock sweep body skips the already-committed step (flag threaded through StepData) and the update path's decrease reserve gains a checked_sub defense documented as unreachable once the structural ordering holds - the old order Arc is moved out with mem::replace in update_entry and match_front's in-place / re-key arms and dropped only after the shard guard releases, so a last-reference deallocation never runs inside the critical section - the update-vs-replenish race test now asserts strictly non-increasing hidden depth while the maker rests (the pre-#115 stale write-back fails it), guarantees matcher progress, and runs the drain across the whole race window --- src/price_level/level.rs | 158 +++++++++++++++++++++------------ src/price_level/order_queue.rs | 45 +++++++--- src/price_level/tests/level.rs | 62 +++++++++---- 3 files changed, 175 insertions(+), 90 deletions(-) diff --git a/src/price_level/level.rs b/src/price_level/level.rs index 2824e1d..dc9a84a 100644 --- a/src/price_level/level.rs +++ b/src/price_level/level.rs @@ -1088,6 +1088,11 @@ impl PriceLevel { hidden_stranded: u64, /// The taker's remaining quantity after this maker is matched. new_remaining: u64, + /// `true` when this step's level-counter deltas were already applied + /// INSIDE the locked decision closure (the replenish path, issue + /// #128). The post-lock body then skips re-applying them so the + /// counters move exactly once. + counters_committed: bool, } // Either the maker progressed (carrying `StepData`), was parked @@ -1195,52 +1200,51 @@ impl PriceLevel { 0 }; - let data = StepData { - consumed, - hidden_reduced, - fully_consumed: updated_order.is_none(), - maker_id, - maker_side, - maker_price, - maker_timestamp, - hidden_stranded, - new_remaining, - }; - + let fully_consumed = updated_order.is_none(); + + // Compute the action. For a replenishment, PUBLISH this step's + // level-counter transition HERE — under the maker's entry lock, + // before returning the action (issue #128) — so a concurrent + // `UpdateQuantity` that next locks this same entry observes the + // level counters already consistent with the replenished queue. + // Applying it only after the entry released would leave the + // counter transiently BELOW the queue's visible sum, and the + // update's `old -> new` decrease could then underflow (`0 - 100` + // wrap). `counters_committed` tells the post-lock body to skip + // re-applying this step's deltas so the counters move exactly once. + let mut counters_committed = false; let action = match updated_order { None => FrontAction::Remove, Some(updated) => { if hidden_reduced > 0 { - // Replenishment: a fresh tranche moves from hidden - // into visible, so the level's visible counter takes - // the net delta `hidden_reduced - consumed` (the - // `fetch_sub(consumed)` + `fetch_add(hidden_reduced)` - // applied after this closure returns). Even when every - // resting order's own total fits `u64`, the level's - // visible SUM can exceed `u64::MAX` once hidden depth - // is converted to visible — a state the counter (and - // the true queue visible sum) cannot represent. - // - // Pre-validate that net delta against the LIVE visible - // counter with checked arithmetic BEFORE committing: - // read the counter (the same reserve-before-commit - // pattern used under the entry lock), and if - // `current - consumed + hidden_reduced` would not fit - // `u64`, ABORT. The abort leaves the maker - // byte-identical (`SetAside` mutates nothing), emits - // no trade, and terminates the sweep, so no younger - // maker trades past this FIFO front and the counter - // never wraps. The stuck depth is unreachable until a - // cancel / downsize frees headroom. - let current_visible = self.visible_quantity.load(Ordering::Relaxed); - let fits = current_visible - .checked_sub(consumed) - .and_then(|v| v.checked_add(hidden_reduced)) - .is_some(); - if !fits { + // Replenishment: a fresh tranche moves hidden -> + // visible. Apply the visible NET delta + // (`- consumed + hidden_reduced`) as ONE checked RMW + // plus the hidden decrement, atomically visible before + // the entry lock releases. The checked `fetch_update` + // supersedes the old load-only `fits` pre-check: even + // when every resting order's own total fits `u64`, the + // level's visible SUM can exceed `u64::MAX` once hidden + // depth converts to visible, so a net delta that would + // overflow ABORTS the step (`SetAside` mutates nothing, + // emits no trade, ends the sweep) — no younger maker + // trades past this FIFO front and the counter never + // wraps. The stuck depth is unreachable until a cancel + // / downsize frees headroom. + let net_ok = self + .visible_quantity + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + c.checked_sub(consumed) + .and_then(|v| v.checked_add(hidden_reduced)) + }) + .is_ok(); + if !net_ok { return (FrontAction::SetAside, StepResult::Abort { maker_id }); } - // Replenishment: refreshed tranche loses priority. + self.hidden_quantity + .fetch_sub(hidden_reduced, Ordering::Relaxed); + counters_committed = true; + // Refreshed tranche loses priority. FrontAction::ReplaceAtTail(Arc::new(updated)) } else { // Pure partial fill: keep priority in place. @@ -1249,6 +1253,19 @@ impl PriceLevel { } }; + let data = StepData { + consumed, + hidden_reduced, + fully_consumed, + maker_id, + maker_side, + maker_price, + maker_timestamp, + hidden_stranded, + new_remaining, + counters_committed, + }; + (action, StepResult::Progressed(data)) }); @@ -1314,8 +1331,13 @@ impl PriceLevel { // not this RMW. The delta is keyed off the committed // action so it never double-counts with a concurrent // cancel (which decrements only the residual it removes). - self.visible_quantity - .fetch_sub(data.consumed, Ordering::Relaxed); + // Skipped for a replenish step: its visible net delta + // (already including `- consumed`) was applied under the + // entry lock in the decision closure (issue #128). + if !data.counters_committed { + self.visible_quantity + .fetch_sub(data.consumed, Ordering::Relaxed); + } let trade_id = Id::from_uuid(trade_id_generator.next()); @@ -1364,10 +1386,14 @@ impl PriceLevel { self.hidden_quantity .fetch_sub(data.hidden_stranded, Ordering::Relaxed); } - } else if data.hidden_reduced > 0 { + } else if data.hidden_reduced > 0 && !data.counters_committed { // Replenishment: a fresh tranche moved from hidden into // visible. The maker stayed resident (re-sequenced in // place by `match_front`), so only the counters move. + // As of issue #128 the replenish path commits this + // transition under the entry lock (`counters_committed`), + // so this post-lock branch is now unreachable for it and + // kept only as a defensive no-op. self.hidden_quantity .fetch_sub(data.hidden_reduced, Ordering::Relaxed); self.visible_quantity @@ -1605,28 +1631,42 @@ impl PriceLevel { new_quantity, } => { // Overflow-checked forward reservation of a level counter for a - // component moving `old -> new`. Increase is validated with - // `checked_add` (rejecting before any queue mutation); decrease - // simply subtracts. `Relaxed`: advisory counters (issue #68). + // component moving `old -> new`. BOTH directions use a checked + // `fetch_update` and reject before any queue mutation: an + // increase must not overflow `u64`, and a decrease must not + // underflow it (issue #128 defense). `Relaxed`: advisory counters + // (issue #68). + // + // The underflow guard is a belt-and-suspenders backstop. It is + // unreachable in practice because issue #128's structural fix + // publishes the match sweep's replenish counter transition UNDER + // the maker's entry lock: by the time this `UpdateQuantity` holds + // that same entry lock and reads `old` from the live maker, the + // level counter already includes this maker's full `old` + // contribution, so `counter >= old >= old - new` and the subtract + // cannot go negative. The check simply refuses to wrap if that + // invariant were ever violated, leaving the level untouched. fn reserve( counter: &std::sync::atomic::AtomicU64, old: u64, new: u64, ) -> Result<(), PriceLevelError> { - if new >= old { - counter - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { - c.checked_add(new - old) - }) - .map(|_| ()) - .map_err(|_| PriceLevelError::InvalidOperation { - message: "price level quantity counter overflow on update" - .to_string(), - }) + let result = if new >= old { + let delta = new - old; + counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + c.checked_add(delta) + }) } else { - counter.fetch_sub(old - new, Ordering::Relaxed); - Ok(()) - } + let delta = old - new; + counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + c.checked_sub(delta) + }) + }; + result + .map(|_| ()) + .map_err(|_| PriceLevelError::InvalidOperation { + message: "price level quantity counter overflow on update".to_string(), + }) } // Undo this call's own `old -> new` reservation (commutative with // concurrent deltas — it reverses exactly what it added). diff --git a/src/price_level/order_queue.rs b/src/price_level/order_queue.rs index ef94799..fe06f1b 100644 --- a/src/price_level/order_queue.rs +++ b/src/price_level/order_queue.rs @@ -394,15 +394,22 @@ impl OrderQueue { // needs no lock protection. The other actions commit their // queue mutations here, under the lock, as before. let mut park_seq: Option = None; + // The order swapped OUT of the slot by a partial fill / + // replenish, captured with `mem::replace` and dropped only + // AFTER the entry lock is released (issue #128), so a + // last-reference deallocation never runs under the shard lock. + let mut evicted: 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. + // `drop`), so the deferred `set_aside` insert and the evicted + // order's drop below never run under the shard lock. match &action { FrontAction::Remove => { // Full consume: remove the entry under the lock, then // drop its index entry. A cancel cannot also remove it // (the entry is gone), so no double counter decrement. + // `remove` consumes the guard, releasing the lock + // before the removed value is dropped. let _ = occupied.remove(); self.index.remove(&seq); } @@ -410,7 +417,10 @@ 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. - occupied.get_mut().1 = residual.clone(); + evicted = Some(std::mem::replace( + &mut occupied.get_mut().1, + residual.clone(), + )); drop(occupied); } FrontAction::ReplaceAtTail(refreshed) => { @@ -425,7 +435,7 @@ impl OrderQueue { { let slot = occupied.get_mut(); slot.0 = new_seq; - slot.1 = refreshed.clone(); + evicted = Some(std::mem::replace(&mut slot.1, refreshed.clone())); } // `occupied` still holds the per-entry lock here, so // re-keying the index — a different structure @@ -454,10 +464,12 @@ impl OrderQueue { } // The entry lock is released on every arm above; a - // possibly-allocating scratch-set insert now runs unlocked. + // possibly-allocating scratch-set insert and the evicted + // order's drop now run unlocked. if let Some(seq) = park_seq { set_aside.insert(seq); } + drop(evicted); return FrontOutcome::Matched { result }; } @@ -515,19 +527,25 @@ impl OrderQueue { order_id, "update_entry: the decided order must keep the id it is stored under" ); - let committed = match decision { + // Each arm swaps the new order into the slot with `mem::replace`, + // capturing the OLD `Arc` in `evicted` (issue #128). The old Arc + // is dropped only AFTER the entry lock is released below, so if + // the queue held the last reference, its deallocation never runs + // inside the shard's critical section. + let (committed, evicted) = match decision { UpdateDecision::KeepInPlace(new_order) => { - occupied.get_mut().1 = new_order.clone(); - new_order + let evicted = + std::mem::replace(&mut occupied.get_mut().1, new_order.clone()); + (new_order, evicted) } UpdateDecision::ReplaceAtTail(new_order) => { let new_seq = self.next_seq.fetch_add(1, Ordering::Relaxed); - let old_seq = { + let (old_seq, evicted) = { let slot = occupied.get_mut(); let old_seq = slot.0; slot.0 = new_seq; - slot.1 = new_order.clone(); - old_seq + let evicted = std::mem::replace(&mut slot.1, new_order.clone()); + (old_seq, evicted) }; // Re-key NEW-KEY-FIRST (issue #127): insert the new // sequence before removing the old one, so the id is @@ -538,9 +556,12 @@ impl OrderQueue { // (the stored sequence is already `new_seq`). self.index.insert(new_seq, order_id); self.index.remove(&old_seq); - new_order + (new_order, evicted) } }; + // Release the shard lock, THEN drop the evicted order. + drop(occupied); + drop(evicted); Some(Ok(committed)) } } diff --git a/src/price_level/tests/level.rs b/src/price_level/tests/level.rs index c96a554..16eb00a 100644 --- a/src/price_level/tests/level.rs +++ b/src/price_level/tests/level.rs @@ -7840,21 +7840,30 @@ mod tests { #[test] fn test_update_vs_match_iceberg_stays_consistent() { // An iceberg maker races a matcher (drawing visible, replenishing from - // hidden) against an updater (resizing visible). Invariants: never - // panics; hidden never exceeds its pre-race value (an update preserves - // live hidden, a match only draws it down — it is never resurrected); - // and the level stays consistent (counters == queue == snapshot). + // hidden) against an updater (resizing visible). #115 derives the resized + // order from the LIVE maker under the entry lock, so hidden is NEVER + // resurrected: while the maker rests, its hidden depth is MONOTONICALLY + // NON-INCREASING (a match only draws it down; an update preserves it, + // never restores a stale higher value). The pre-#115 stale-pre-read code + // wrote back a stale hidden after a concurrent match drew it down, + // producing an INCREASE this test's strict monotonic assertion catches. + // + // The matcher runs a FIXED number of matches (so it can never run zero + // times) and signals `done` at the end, while the updater resizes for the + // whole race; we also assert the matcher committed real fills, so the + // race is genuinely exercised rather than trivially satisfied. use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering}; use std::sync::{Arc as StdArc, Barrier}; use std::thread; - const INITIAL_HIDDEN: u64 = 400; - const UPDATE_ROUNDS: usize = 200; + const INITIAL_HIDDEN: u64 = 4_000; + const VISIBLE: u64 = 20; + const MATCH_ITERS: usize = 500; for iter in 0..25 { let level = StdArc::new(PriceLevel::new(10_000)); level - .add_order(create_buy_iceberg_order(1, 10_000, 20, INITIAL_HIDDEN)) + .add_order(create_buy_iceberg_order(1, 10_000, VISIBLE, INITIAL_HIDDEN)) .expect("seed iceberg"); let id = Id::from_u64(1); let barrier = StdArc::new(Barrier::new(2)); @@ -7866,14 +7875,17 @@ mod tests { let done = StdArc::clone(&done); thread::spawn(move || { barrier.wait(); - for r in 0..UPDATE_ROUNDS { - let new_visible = 5 + (r as u64 % 30); + // Resize continuously until the matcher is done, so an update + // races every stage of the drain (not a fixed short burst). + let mut r = 0u64; + while !done.load(AtomicOrdering::Acquire) { + let new_visible = 5 + (r % 30); let _ = level.update_order(OrderUpdate::UpdateQuantity { order_id: id, new_quantity: Quantity::new(new_visible), }); + r += 1; } - done.store(true, AtomicOrdering::Release); }) }; let matcher = { @@ -7883,9 +7895,10 @@ mod tests { thread::spawn(move || { barrier.wait(); let generator = UuidGenerator::new(Uuid::from_u128(0xF00D_0000 + iter as u128)); - let mut max_hidden = 0u64; - while !done.load(AtomicOrdering::Acquire) { - let _ = level.match_order( + let mut committed = 0usize; + let mut prev_hidden = u64::MAX; + for _ in 0..MATCH_ITERS { + let result = level.match_order( 3, Id::from_u64(999), TimeInForce::Gtc, @@ -7893,25 +7906,36 @@ mod tests { TimestampMs::new(1_700_000_000_000), &generator, ); - // Sample the resting maker's hidden depth (if any). + if result.executed_quantity().map(|q| q.as_u64()).unwrap_or(0) > 0 { + committed += 1; + } + // Sample the resting maker's hidden depth; it must never + // rise across the race (strict, tight enough that a + // pre-#115 resurrection would fail here). if let Some(o) = level .snapshot_by_insertion_seq() .into_iter() .find(|o| o.id() == id) { - max_hidden = max_hidden.max(o.hidden_quantity().as_u64()); + let h = o.hidden_quantity().as_u64(); + assert!( + h <= prev_hidden, + "iter {iter}: hidden rose {prev_hidden} -> {h} (resurrection; #115 regression)" + ); + prev_hidden = h; } } - max_hidden + done.store(true, AtomicOrdering::Release); + committed }) }; + let committed = matcher.join().expect("matcher panicked"); updater.join().expect("updater panicked"); - let max_hidden = matcher.join().expect("matcher panicked"); assert!( - max_hidden <= INITIAL_HIDDEN, - "iter {iter}: hidden {max_hidden} exceeded its pre-race value {INITIAL_HIDDEN} (resurrection)" + committed >= 1, + "iter {iter}: matcher committed no fills — the race was not exercised" ); assert_counters_match_queue(&level); }