pricelevel 0.9.1

use crate::errors::PriceLevelError;
use crate::orders::{Id, OrderType};
use crossbeam_skiplist::SkipMap;
use dashmap::DashMap;
use dashmap::mapref::entry::Entry;
use serde::de::{SeqAccess, Visitor};
use serde::ser::SerializeSeq;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::collections::HashSet;
use std::fmt;
use std::fmt::Display;
use std::marker::PhantomData;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

/// A thread-safe queue of orders with specialized operations.
///
/// Time priority (price-time / FIFO within the level) is maintained by an
/// ordered index keyed by a monotonic insertion sequence rather than a plain
/// tail-only FIFO. This lets a partially-filled maker keep its place at the
/// front of the queue: the residual is re-inserted at its *original* sequence,
/// instead of being appended to the tail.
#[derive(Debug)]
pub struct OrderQueue {
    /// A map of order IDs to `(insertion sequence, order)` for O(1) lookups.
    /// The sequence travels with the value so it can be recovered on pop and
    /// reused when re-inserting a partial-fill residual.
    orders: DashMap<Id, (u64, Arc<OrderType<()>>)>,
    /// Ordered index `sequence -> Id`. The lowest sequence is the front
    /// (oldest) order, so iteration / pop honours strict time priority.
    index: SkipMap<u64, Id>,
    /// Monotonic source of insertion sequences.
    next_seq: AtomicU64,
}

/// The mutation a matcher decides to apply to the front maker it is currently
/// matching, while the maker's `orders` entry is held under the per-entry lock.
///
/// Returned by the decision closure passed to [`OrderQueue::match_front`]. The
/// queue applies the variant atomically (under the same per-entry lock that
/// guards a concurrent [`OrderQueue::remove`]), so a `cancel` of the same id
/// either runs entirely before the decision (the closure observes `Vacant` and
/// is never called) or entirely after the commit (it observes the residual /
/// emptiness the matcher left behind). A cancel can never be lost mid-decision.
#[derive(Debug)]
pub(crate) enum FrontAction {
    /// The maker was fully consumed: remove it from `orders` and drop its index
    /// entry. After this the id no longer rests at the level.
    Remove,
    /// Pure partial fill: keep the maker at its current insertion sequence
    /// (and therefore its price-time / FIFO position) by swapping the stored
    /// value to the residual in place under the per-entry lock.
    KeepInPlace(Arc<OrderType<()>>),
    /// Iceberg / reserve replenishment: the refreshed tranche loses time
    /// priority, so remove the old entry and re-queue the new order at the tail
    /// with a fresh insertion sequence.
    ReplaceAtTail(Arc<OrderType<()>>),
    /// The maker made no progress this sweep (a degenerate zero-progress shape).
    /// Leave it untouched in `orders`/`index`; the caller sets its sequence
    /// aside so the sweep advances to the maker behind it without re-popping it.
    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<OrderType<()>>),
    /// 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<OrderType<()>>),
}

/// 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)]
pub(crate) enum FrontOutcome<R> {
    /// A front candidate existed and the decision closure ran. Carries the
    /// closure's result `R` (the trade bookkeeping data the sweep needs to apply
    /// counter deltas and emit the trade). The committed [`FrontAction`] is
    /// already encoded in that bookkeeping (full consume vs partial vs
    /// replenish), so it is not surfaced separately.
    Matched { result: R },
    /// The queue is empty (no front candidate that is not already set aside).
    /// The sweep is done.
    Empty,
}

impl OrderQueue {
    /// Create a new empty order queue
    #[must_use]
    pub fn new() -> Self {
        Self {
            orders: DashMap::new(),
            index: SkipMap::new(),
            next_seq: AtomicU64::new(0),
        }
    }

    /// Add an order to the tail of the queue (newest time priority),
    /// **unconditionally overwriting** any existing entry for the same id.
    ///
    /// 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<OrderType<()>>) {
        // `Relaxed` is sufficient: only the uniqueness and monotonicity of the
        // counter matter. The happens-before ordering between concurrent
        // producers/consumers is provided by the lock-free `index`/`orders`
        // structures, not by this counter, so no synchronization rides on it.
        let seq = self.next_seq.fetch_add(1, Ordering::Relaxed);
        let order_id = order.id();
        self.orders.insert(order_id, (seq, order));
        self.index.insert(seq, order_id);
    }

    /// Insert an order only if its id is not already present — the admission
    /// primitive.
    ///
    /// Thin wrapper over the crate-internal `try_push_with` with a no-op
    /// reservation: the id-uniqueness, held-lock publication, and
    /// no-sequence-gap guarantees documented there all apply. Use this when
    /// 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
    ///
    /// Returns [`PriceLevelError::DuplicateOrderId`] if an order with the same
    /// id already rests in the queue.
    #[must_use = "a rejected duplicate must be handled, not ignored"]
    pub fn try_push(&self, order: Arc<OrderType<()>>) -> Result<(), PriceLevelError> {
        self.try_push_with(order, || Ok(()))
    }

    /// Insert an order only if its id is absent, committing a caller-supplied
    /// `reserve` step **atomically with the publication** — the admission
    /// primitive with a reservation hook.
    ///
    /// The `DashMap` entry API makes the whole operation atomic under the
    /// per-shard lock:
    ///
    /// 1. **Identity is decided first.** An `Occupied` entry means the id
    ///    already rests here: return [`PriceLevelError::DuplicateOrderId`]
    ///    with **nothing touched** — `reserve` is never run, no sequence is
    ///    minted, no counter moves. This is why a duplicate can never leave a
    ///    transient side effect (e.g. an inflated level counter) for another
    ///    thread to observe, and why a duplicate at counter capacity reports
    ///    `DuplicateOrderId` rather than a spurious overflow.
    /// 2. **Then `reserve` runs**, still under the shard lock, now that the id
    ///    is known free. If it returns `Err`, propagate it with nothing
    ///    inserted — the caller is responsible for leaving its own state
    ///    unchanged on `Err` (e.g. rolling back a partial multi-counter
    ///    reservation before returning).
    /// 3. **Then publish, holding the shard lock across the index insert.** The
    ///    map value is inserted (its returned guard keeps the shard write lock),
    ///    the `seq -> id` index entry is added while that guard is still held,
    ///    and only then is the guard dropped. Holding the lock across both
    ///    publications closes the window where the lock released after the map
    ///    insert but before the index insert: a concurrent cancel + same-id
    ///    readmission could otherwise land its own entries and let this call's
    ///    stale index insert produce two index entries for one id. This is the
    ///    same held-lock shape [`OrderQueue::match_front`]'s `ReplaceAtTail`
    ///    uses. `self.index` is a separate structure (`SkipMap`), so inserting
    ///    into it under the `DashMap` shard lock cannot deadlock.
    ///
    /// The insertion sequence is minted **inside** the `Vacant` arm, after
    /// `reserve` succeeds, so neither a rejected duplicate nor a failed
    /// reservation consumes a sequence (no gaps) and the index entry is added
    /// only for the order that actually landed in the map.
    ///
    /// `reserve` runs while the shard lock is held, so it MUST NOT call back
    /// into this queue (that would deadlock on the same shard) and MUST NOT
    /// block; the level's counter reservations (plain atomic RMWs) satisfy both.
    ///
    /// # Errors
    ///
    /// Returns [`PriceLevelError::DuplicateOrderId`] if an order with the same
    /// id already rests in the queue, or whatever error `reserve` returns.
    #[must_use = "a rejected admission must be handled, not ignored"]
    pub(crate) fn try_push_with<F>(
        &self,
        order: Arc<OrderType<()>>,
        reserve: F,
    ) -> Result<(), PriceLevelError>
    where
        F: FnOnce() -> Result<(), PriceLevelError>,
    {
        let order_id = order.id();
        match self.orders.entry(order_id) {
            Entry::Occupied(_) => Err(PriceLevelError::DuplicateOrderId(order_id.to_string())),
            Entry::Vacant(slot) => {
                // Identity is already decided (this arm means the id is free).
                // Run the caller's reservation before publishing; on failure
                // nothing has been inserted and no sequence minted, so the
                // level stays byte-identical once the caller unwinds its own
                // partial reservation.
                reserve()?;
                // Mint the sequence only now that the id is free and the
                // reservation committed, so neither a rejected duplicate nor a
                // failed reservation leaves a gap in the sequence.
                let seq = self.next_seq.fetch_add(1, Ordering::Relaxed);
                // Hold the shard lock across BOTH publications: the map insert
                // returns a guard that keeps the lock, the index entry is added
                // while it is held, and only then is the guard dropped.
                let guard = slot.insert((seq, order));
                self.index.insert(seq, order_id);
                drop(guard);
                Ok(())
            }
        }
    }

    /// Pop the front (oldest) order together with its insertion sequence,
    /// removing it from the queue.
    ///
    /// The sequence is returned alongside the order. As of #81 the match sweep
    /// no longer pops-then-reinserts a maker (it operates in place under the
    /// per-entry lock via [`OrderQueue::match_front`]); this remains the backing
    /// of the destructive [`OrderQueue::pop`] used by tests and queue draining.
    #[must_use]
    pub(crate) fn pop_entry(&self) -> Option<(u64, Arc<OrderType<()>>)> {
        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();
            // 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,
            }
        }
    }

    /// Attempt to pop an order from the queue (front / oldest first).
    #[must_use]
    pub fn pop(&self) -> Option<Arc<OrderType<()>>> {
        self.pop_entry().map(|(_, order)| order)
    }

    /// Select the front (oldest, not-yet-set-aside) maker and apply a match
    /// decision to it **atomically with respect to a concurrent
    /// [`OrderQueue::remove`] (cancel) of the same id**.
    ///
    /// This replaces the old `pop_entry` + later `reinsert` sequence used by the
    /// match sweep, which removed the maker from `orders` before deciding and so
    /// opened a "lost cancel" window: a `cancel` landing between the pop and the
    /// reinsert would `remove` an id that was no longer in `orders`, silently
    /// no-op (no counter decrement), and the matcher would then reinsert the
    /// residual — leaving the cancelled order resting.
    ///
    /// Here the maker is **kept resident in `orders`** while it is matched. The
    /// decision and the resulting mutation both run while the maker's `orders`
    /// entry is held under DashMap's per-entry (shard) lock — the same lock a
    /// concurrent `cancel`'s [`DashMap::remove`] must take. The two therefore
    /// serialize on that lock:
    ///
    /// - If `cancel` wins the lock first, this method observes the entry as
    ///   `Vacant` (cancel already removed it + decremented counters), drops the
    ///   stale index entry, and advances to the next candidate. The decision
    ///   closure is never run for the cancelled id.
    /// - If this method wins, it commits its [`FrontAction`] under the lock; a
    ///   `cancel` arriving afterwards observes whatever the matcher left — the
    ///   residual for a partial fill (and removes *that*, decrementing by the
    ///   residual), or nothing for a full consume (and correctly no-ops).
    ///
    /// In every interleaving a cancel either fully wins or fully loses; it is
    /// never lost. The counter delta for the matched transition is the caller's
    /// responsibility and is keyed off the returned [`FrontAction`], so it can
    /// never double-count with the cancel.
    ///
    /// `set_aside` carries the insertion sequences of makers the current sweep
    /// has parked (the no-progress guard); they are skipped when choosing the
    /// front so the sweep does not re-pick them. A `SetAside` action inserts the
    /// chosen seq into it. It is a `HashSet` so membership during the front scan
    /// is O(1) rather than the O(n) `Vec::contains` it replaced; it is only ever
    /// inserted into and probed, never iterated for ordering.
    ///
    /// `decide` is the pure match decision (e.g. [`OrderType::match_against`]
    /// plus trade bookkeeping). It runs while the per-entry lock is held, so it
    /// MUST NOT call back into this queue (that would deadlock on the same
    /// shard) and MUST NOT block. It receives the maker's insertion `seq` and an
    /// immutable borrow of the resident order; the borrow ends before any commit
    /// mutates the entry, so the decision must return OWNED action data and no
    /// reference may escape it.
    ///
    /// All three mutating actions keep the maker's value **resident in `orders`
    /// until it is genuinely gone** (a full consume) — even the
    /// [`FrontAction::ReplaceAtTail`] re-prioritisation swaps the value and
    /// re-sequences it in place rather than removing-then-re-pushing — so the
    /// lost-cancel window is closed for every action, not just the partial fill.
    pub(crate) fn match_front<F, R>(
        &self,
        set_aside: &mut HashSet<u64>,
        decide: F,
    ) -> FrontOutcome<R>
    where
        F: FnOnce(u64, &OrderType<()>) -> (FrontAction, R),
    {
        loop {
            // Find the lowest-sequence index entry not already set aside this
            // sweep. `index.iter()` yields entries in ascending sequence order
            // (front = oldest = highest time priority).
            let Some((seq, order_id)) = self
                .index
                .iter()
                .find(|e| !set_aside.contains(e.key()))
                .map(|e| (*e.key(), *e.value()))
            else {
                return FrontOutcome::Empty;
            };

            // Lock the maker's `orders` entry. `entry` takes the shard write
            // lock, which a concurrent `cancel`'s `remove` must also take, so the
            // decision + mutation below are atomic with respect to that cancel.
            match self.orders.entry(order_id) {
                Entry::Vacant(_) => {
                    // The maker was cancelled (removed from `orders`) but its
                    // index entry is stale. Drop the stale index entry and retry
                    // with the next front candidate. The cancel already
                    // decremented the counters, so there is nothing to account
                    // here.
                    self.index.remove(&seq);
                    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
                    // path: the immutable borrow lives only for the `decide`
                    // call, which returns OWNED action data, so it ends before
                    // any `get_mut()` / `remove()` commit below (no reference
                    // escapes into a `FrontAction`).
                    let (action, result) = decide(seq, occupied.get().1.as_ref());

                    // A `SetAside` records a sequence into the caller's scratch
                    // `HashSet`, whose first insert allocates. Defer that insert
                    // until AFTER the entry lock is released (issue #126) so no
                    // allocation ever runs under the shard lock — the set is
                    // per-sweep scratch owned by the caller, never shared, so it
                    // needs no lock protection. The other actions commit their
                    // queue mutations here, under the lock, as before.
                    let mut park_seq: Option<u64> = 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<Arc<OrderType<()>>> = 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 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);
                        }
                        FrontAction::KeepInPlace(residual) => {
                            // Partial fill keeping priority: swap the stored value
                            // to the residual in place, keeping the same
                            // sequence/index entry. Still under the entry lock.
                            evicted = Some(std::mem::replace(
                                &mut occupied.get_mut().1,
                                residual.clone(),
                            ));
                            drop(occupied);
                        }
                        FrontAction::ReplaceAtTail(refreshed) => {
                            // Replenished tranche loses time priority, but the
                            // maker keeps the SAME id and must stay resident in
                            // `orders` so a concurrent cancel cannot slip into a
                            // remove-then-push gap. So: mint a fresh tail sequence
                            // and swap BOTH the value and its stored sequence in
                            // place under the entry lock; only the index is
                            // re-keyed (old seq -> new seq) afterwards.
                            let new_seq = self.next_seq.fetch_add(1, Ordering::Relaxed);
                            {
                                let slot = occupied.get_mut();
                                slot.0 = new_seq;
                                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. 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
                            // stale `index[seq|new_seq] -> id` entry pointing at an
                            // 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.insert(new_seq, order_id);
                            self.index.remove(&seq);
                            drop(occupied);
                        }
                        FrontAction::SetAside => {
                            // No progress: leave the entry untouched. Release the
                            // lock and park its sequence below.
                            drop(occupied);
                            park_seq = Some(seq);
                        }
                    }

                    // The entry lock is released on every arm above; a
                    // possibly-allocating scratch-set insert 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 };
                }
            }
        }
    }

    /// 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<F>(
        &self,
        order_id: Id,
        decide: F,
    ) -> Option<Result<Arc<OrderType<()>>, PriceLevelError>>
    where
        F: FnOnce(&OrderType<()>) -> Result<UpdateDecision, PriceLevelError>,
    {
        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
    /// the queue, keeping its time priority ahead of orders that arrived later.
    ///
    /// As of #81 the match sweep no longer removes-then-reinserts a partially
    /// filled maker; it swaps the residual in place under the per-entry lock via
    /// [`OrderQueue::match_front`], which keeps the maker resident in `orders`
    /// the whole time (closing the lost-cancel window). This helper survives
    /// only as a queue-priority test fixture and is therefore `#[cfg(test)]`.
    #[cfg(test)]
    pub(crate) fn reinsert(&self, seq: u64, order: Arc<OrderType<()>>) {
        let order_id = order.id();
        self.orders.insert(order_id, (seq, order));
        self.index.insert(seq, order_id);
    }

    /// Search for an order with the given ID. O(1) operation.
    #[must_use]
    #[inline]
    pub fn find(&self, order_id: Id) -> Option<Arc<OrderType<()>>> {
        self.orders.get(&order_id).map(|o| o.value().1.clone())
    }

    /// Remove an order with the given ID.
    /// Returns the removed order if found. Cleans both the map and the index.
    #[must_use]
    pub fn remove(&self, order_id: Id) -> Option<Arc<OrderType<()>>> {
        let (_, (seq, order)) = self.orders.remove(&order_id)?;
        self.index.remove(&seq);
        Some(order)
    }

    /// Test-only invariant check: the id-keyed map and the ordered index are
    /// **1:1**. Must be called only at quiescence (no concurrent mutation), when
    /// every in-flight operation has completed.
    ///
    /// Verifies there are exactly as many index entries as map entries and that
    /// every index entry `seq -> id` points to a map entry whose stored sequence
    /// is exactly `seq`. A split index entry (two sequences for one id — the
    /// publication-race bug closed by [`OrderQueue::try_push_with`] holding the
    /// shard lock across both publications) makes the index longer than the map,
    /// so this returns `false`.
    #[cfg(test)]
    #[must_use]
    pub(crate) fn debug_map_index_consistent(&self) -> bool {
        if self.index.len() != self.orders.len() {
            return false;
        }
        self.index.iter().all(|entry| {
            let seq = *entry.key();
            let id = *entry.value();
            self.orders
                .get(&id)
                .is_some_and(|slot| slot.value().0 == seq)
        })
    }

    /// Iterate through current orders without materializing an intermediate vector.
    pub fn iter_orders(&self) -> impl Iterator<Item = Arc<OrderType<()>>> + '_ {
        self.orders.iter().map(|entry| entry.value().1.clone())
    }

    /// Materialize a stable snapshot vector sorted by `(timestamp, sequence)`.
    ///
    /// The insertion sequence is used as a deterministic tiebreak so orders
    /// sharing a millisecond timestamp are still ordered exactly as matching
    /// would consume them. This is the timestamp-sorted display / reporting
    /// view; it is not what a snapshot round-trip uses. Snapshot round-trips
    /// materialize via `snapshot_by_seq` (ascending insertion sequence), so the
    /// live queue order — including the "sizing up loses time priority"
    /// demotion — survives a restore.
    #[must_use]
    pub fn snapshot_vec(&self) -> Vec<Arc<OrderType<()>>> {
        let mut orders: Vec<(u64, Arc<OrderType<()>>)> =
            self.orders.iter().map(|o| o.value().clone()).collect();
        orders.sort_by_key(|(seq, o)| (o.timestamp(), *seq));
        orders.into_iter().map(|(_, o)| o).collect()
    }

    /// Convert the queue to a vector (for compatibility and snapshots).
    #[must_use]
    pub fn to_vec(&self) -> Vec<Arc<OrderType<()>>> {
        self.snapshot_vec()
    }

    /// Materialize the resting orders in ascending **insertion-sequence** order —
    /// the exact order [`OrderQueue::match_front`] consumes them.
    ///
    /// Derived from the authoritative `orders` map (keyed by `Id`) rather than
    /// the `index`: we collect the `(stored_seq, order)` pairs and sort by the
    /// 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 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
    /// `index` instead could pin a stale `seq -> id` entry during a
    /// re-sequencing and emit the same order twice (or at the wrong priority),
    /// which corrupts a snapshot fold.
    ///
    /// Unlike [`OrderQueue::snapshot_vec`] (sorted by `(timestamp, sequence)`),
    /// this reflects pure insertion order, so it equals the sweep even when
    /// timestamps are not monotonic with insertion.
    ///
    /// This view also backs [`crate::price_level::PriceLevel::snapshot`]: the
    /// snapshot round-trip re-enqueues in this consumption order, so exact
    /// price-time priority — including the "sizing up loses time priority"
    /// demotion — is preserved across a restore.
    #[must_use]
    pub(crate) fn snapshot_by_seq(&self) -> Vec<Arc<OrderType<()>>> {
        let mut out = Vec::new();
        self.snapshot_by_seq_into(&mut out);
        out
    }

    /// Fill `out` with the resting orders in ascending **insertion-sequence**
    /// order — the buffer-reuse variant of [`OrderQueue::snapshot_by_seq`].
    ///
    /// `out` is cleared first, then extended in place, so a caller can reuse one
    /// scratch buffer across calls and avoid the per-call allocation of the
    /// returned `Vec`. Note the internal `(seq, order)` pairs buffer plus its
    /// sort is still paid on every call — the reuse saves only the output `Vec`
    /// allocation, not the collect-and-sort. The duplicate-free,
    /// committed-pair guarantees are identical to
    /// [`OrderQueue::snapshot_by_seq`]; the only difference is where the result
    /// lands.
    pub(crate) fn snapshot_by_seq_into(&self, out: &mut Vec<Arc<OrderType<()>>>) {
        // Build from the `orders` map (one entry per id) so a concurrent
        // re-sequencing can never surface an order twice or at a mixed
        // priority; see `snapshot_by_seq` for the full rationale.
        let mut pairs: Vec<(u64, Arc<OrderType<()>>)> = self
            .orders
            .iter()
            .map(|entry| entry.value().clone())
            .collect();
        // Unstable sort is deterministic here because sequences are unique
        // 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));
    }

    /// Creates a new `OrderQueue` instance and populates it with orders from the provided vector.
    ///
    /// This function takes ownership of a vector of order references (wrapped in `Arc`) and constructs
    /// a new `OrderQueue` by iteratively pushing each order into the queue. The resulting queue
    /// maintains the insertion order of the original vector.
    ///
    /// # Parameters
    ///
    /// * `orders` - A vector of atomic reference counted (`Arc`) order instances representing
    ///   the orders to be added to the new queue.
    ///
    /// # Returns
    ///
    /// A new `OrderQueue` instance containing all the orders from the input vector.
    ///
    /// Note: this infallible constructor drops later orders that repeat an id
    /// already inserted (keep-first), so the queue's id-keyed map and its
    /// ordered index always stay 1:1. Callers that must *reject* a
    /// duplicate-bearing vector (e.g. snapshot restore) validate uniqueness
    /// upstream where a `Result` can be returned.
    ///
    /// `pub(crate)`: a keep-first constructor that silently drops duplicates is
    /// not a safe public entry point (a caller could restore counters computed
    /// over a copy the queue then discards). The public path is
    /// [`PriceLevel::from_snapshot`](crate::price_level::PriceLevel), which
    /// rejects duplicates; this stays crate-internal for tests and the
    /// upstream-validated restore.
    #[allow(dead_code)]
    #[must_use]
    pub(crate) fn from_vec(orders: Vec<Arc<OrderType<()>>>) -> Self {
        let queue = OrderQueue::new();
        for order in orders {
            // Keep-first on a duplicate id: the index must stay 1:1.
            let _ = queue.try_push(order);
        }
        queue
    }

    /// Check if the queue is empty
    #[allow(dead_code)]
    #[must_use]
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.orders.is_empty()
    }

    /// Returns the number of orders currently in the queue.
    ///
    /// # Returns
    ///
    /// * `usize` - The total count of orders in the queue.
    ///
    #[must_use]
    #[inline]
    pub fn len(&self) -> usize {
        self.orders.len()
    }
}

impl Default for OrderQueue {
    fn default() -> Self {
        Self::new()
    }
}
// Implement serialization for OrderQueue
impl Serialize for OrderQueue {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        // Materialize the ordered view first so the length hint always matches
        // the number of elements emitted. `snapshot_by_seq` derives its pairs
        // from the `orders` map (one entry per id) and sorts by insertion
        // sequence, so it can neither duplicate an order nor disagree with its
        // own length during a concurrent re-sequencing. Insertion-sequence
        // order keeps the round-trip price-time priority (the DashMap alone has
        // no deterministic iteration order).
        let ordered = self.snapshot_by_seq();
        let mut seq = serializer.serialize_seq(Some(ordered.len()))?;
        for order in &ordered {
            seq.serialize_element(order.as_ref())?;
        }
        seq.end()
    }
}

impl FromStr for OrderQueue {
    type Err = PriceLevelError;
    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if !s.starts_with("OrderQueue:orders=[") || !s.ends_with(']') {
            return Err(PriceLevelError::ParseError {
                message: "Invalid format".to_string(),
            });
        }

        let content = &s["OrderQueue:orders=[".len()..s.len() - 1];
        let queue = OrderQueue::new();

        if !content.is_empty() {
            for order_str in content.split(',') {
                let order =
                    OrderType::from_str(order_str).map_err(|e| PriceLevelError::ParseError {
                        message: format!("Order parse error: {e}"),
                    })?;
                // Reject a repeated id rather than overwriting it.
                queue.try_push(Arc::new(order))?;
            }
        }

        Ok(queue)
    }
}

impl Display for OrderQueue {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "OrderQueue:orders=[")?;
        let mut first = true;
        for order in self.snapshot_vec() {
            if !first {
                write!(f, ",")?;
            }
            write!(f, "{order}")?;
            first = false;
        }
        write!(f, "]")
    }
}

impl From<Vec<Arc<OrderType<()>>>> for OrderQueue {
    /// Infallible conversion: a repeated id is dropped (keep-first) so the map
    /// and index stay 1:1. Restore paths that must reject duplicates validate
    /// uniqueness upstream (see [`crate::price_level::PriceLevel::from_snapshot`]).
    fn from(orders: Vec<Arc<OrderType<()>>>) -> Self {
        let queue = OrderQueue::new();
        for order in orders {
            let _ = queue.try_push(order);
        }
        queue
    }
}

// Custom visitor for deserializing OrderQueue
struct OrderQueueVisitor {
    marker: PhantomData<fn() -> OrderQueue>,
}

impl OrderQueueVisitor {
    fn new() -> Self {
        OrderQueueVisitor {
            marker: PhantomData,
        }
    }
}

impl<'de> Visitor<'de> for OrderQueueVisitor {
    type Value = OrderQueue;

    fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
        formatter.write_str("a sequence of orders")
    }

    fn visit_seq<V>(self, mut seq: V) -> Result<OrderQueue, V::Error>
    where
        V: SeqAccess<'de>,
    {
        let queue = OrderQueue::new();

        // Deserialize each order and add it to the queue, rejecting a repeated
        // id rather than silently overwriting it.
        while let Some(order) = seq.next_element::<OrderType<()>>()? {
            queue
                .try_push(Arc::new(order))
                .map_err(serde::de::Error::custom)?;
        }

        Ok(queue)
    }
}

// Implement deserialization for OrderQueue
impl<'de> Deserialize<'de> for OrderQueue {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        // Deserialize as a sequence of orders
        deserializer.deserialize_seq(OrderQueueVisitor::new())

        // Alternative approach: Deserialize as OrderQueueData first, then convert
        // let data = OrderQueueData::deserialize(deserializer)?;
        // let queue = OrderQueue::new();
        // for order in data.orders {
        //     queue.push(Arc::new(order));
        // }
        // Ok(queue)
    }
}