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..dc9a84a 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}; @@ -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 @@ -1087,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 @@ -1194,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. @@ -1248,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)) }); @@ -1313,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()); @@ -1363,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 @@ -1531,14 +1558,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, @@ -1590,117 +1630,110 @@ 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(), - } - })?; - - // 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. + // Overflow-checked forward reservation of a level counter for a + // 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). // - // 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); + // 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> { + let result = if new >= old { + let delta = new - old; + counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| { + c.checked_add(delta) + }) + } else { + 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). + 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 (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 { - return Ok(None); // Removed by another thread. - }; - self.orders.push(new_order_arc.clone()); - removed - } 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` / `remove` + `push`) - // 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 e949bc2..fe06f1b 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)] @@ -91,18 +107,15 @@ 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 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 // counter matter. The happens-before ordering between concurrent @@ -123,6 +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-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 /// @@ -223,11 +240,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, } } } @@ -324,6 +356,27 @@ 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 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 + // 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 @@ -341,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); } @@ -357,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) => { @@ -372,12 +435,15 @@ 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 // (`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 @@ -385,8 +451,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 => { @@ -398,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 }; } @@ -409,6 +477,96 @@ impl OrderQueue { } } + /// 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, + decide: F, + ) -> Option>, PriceLevelError>> + where + F: FnOnce(&OrderType<()>) -> Result, + { + match self.orders.entry(order_id) { + Entry::Vacant(_) => None, + Entry::Occupied(mut occupied) => { + // 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" + ); + // 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) => { + 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, evicted) = { + let slot = occupied.get_mut(); + let old_seq = slot.0; + slot.0 = new_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 + // 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, evicted) + } + }; + // Release the shard lock, THEN drop the evicted order. + drop(occupied); + drop(evicted); + Some(Ok(committed)) + } + } + } + /// 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 @@ -433,35 +591,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] @@ -532,8 +661,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 @@ -577,8 +706,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 a08f7e0..16eb00a 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}; @@ -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] @@ -7343,6 +7364,582 @@ 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}" + ); + } + } + } + + // ------------------------------------------------------------------ + // 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). #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 = 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, VISIBLE, 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(); + // 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; + } + }) + }; + 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 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, + TakerKind::Standard, + TimestampMs::new(1_700_000_000_000), + &generator, + ); + 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) + { + let h = o.hidden_quantity().as_u64(); + assert!( + h <= prev_hidden, + "iter {iter}: hidden rose {prev_hidden} -> {h} (resurrection; #115 regression)" + ); + prev_hidden = h; + } + } + done.store(true, AtomicOrdering::Release); + committed + }) + }; + + let committed = matcher.join().expect("matcher panicked"); + updater.join().expect("updater panicked"); + + assert!( + committed >= 1, + "iter {iter}: matcher committed no fills — the race was not exercised" + ); + 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 061ab65..22818f9 100644 --- a/src/price_level/tests/order_queue.rs +++ b/src/price_level/tests/order_queue.rs @@ -610,4 +610,164 @@ 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, UpdateDecision}; + 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.update_entry(Id::from_u64(1), |_live| { + Ok(UpdateDecision::ReplaceAtTail(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. + 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.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()); + + 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() { + 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), + // 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.update_entry(Id::from_u64(N / 2), |_live| { + Ok(UpdateDecision::ReplaceAtTail(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" + ); + } + } }