rustrade-execution 0.17.0

//! Learning how orders ended while an account stream was disconnected.
//!
//! A reconnect recovers the fills it missed by time, but an order cancelled, expired or rejected in
//! the meantime cannot be found that way: Binance and Alpaca both filter their order history on
//! when an order was created, so a window opening at the disconnect misses an order placed before
//! it. So the client keeps the orders it knows to be live ([`KnownLiveOrders`]), and after a
//! reconnect lists the open orders of each instrument it holds one on and asks about each held
//! order no longer listed, by client order id ([`recover_ended_orders`]). [`UncheckedOrders`] holds
//! which instruments still need that check, and when a failed one is retried. [`fetch_ended_by_key`]
//! answers [`OrderStatusClient`](crate::client::OrderStatusClient) with the same lookup.
//!
//! Nothing here names a venue's endpoint: each client passes in its own listing and lookup, and
//! says through [`PendingFills`] which instruments still wait for fills it has to recover.

use crate::{
    AccountEventKind, UnindexedAccountEvent,
    error::{ApiError, UnindexedClientError},
    order::{
        OrderKind, UnindexedInactiveOrder, UnindexedOrderKey,
        id::{ClientOrderId, OrderId, StrategyId},
        state::{ActiveOrderState, Open, OrderState, UnindexedOrderState},
    },
};
use fnv::{FnvHashMap, FnvHashSet};
use lru::LruCache;
use rust_decimal::Decimal;
use rustrade_instrument::{exchange::ExchangeId, instrument::name::InstrumentNameExchange};
use rustrade_integration::collection::snapshot::Snapshot;
use std::{
    collections::BTreeMap,
    future::Future,
    num::NonZeroUsize,
    ops::{Deref, DerefMut},
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
    },
    time::Duration,
};
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};

/// How many times a failed read is retried after its first failure before it is given up: a
/// reconnect's check of how orders ended here, and a Binance fill gap.
pub(crate) const MAX_GAP_RETRIES: u32 = 5;
/// The wait before a failed read's first retry. It doubles after each failure, so the retries come
/// 1, 2, 4, 8 and 16 minutes apart, about half an hour in all.
pub(crate) const GAP_RETRY_BASE_SECS: u64 = 60;

/// What became of a read that failed or did not finish: a reconnect's check of how orders ended,
/// or a Binance fill gap.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum GapFailure {
    /// It is read again after this delay.
    Retry(Duration),
    /// It has failed [`MAX_GAP_RETRIES`] retries and is dropped.
    GivenUp,
}

/// Which instruments still have fills a reconnect has to recover, so that how an order ended is
/// not reported before its fills.
pub(crate) trait PendingFills {
    /// Whether fills on `instrument` are still to be recovered.
    fn pending(&self, instrument: &InstrumentNameExchange) -> bool;
}

/// No fills left to recover, as for a client whose fill recovery is done, or given up, before its
/// order check starts.
#[derive(Debug, Clone, Copy, Default)]
// Alpaca and IBKR recover their fills before their checks, and Hyperliquid has them redelivered
// before its own.
#[cfg_attr(
    not(any(feature = "alpaca", feature = "hyperliquid", feature = "ibkr")),
    allow(dead_code)
)]
pub(crate) struct NoPendingFills;

impl PendingFills for NoPendingFills {
    fn pending(&self, _: &InstrumentNameExchange) -> bool {
        false
    }
}

/// How many live orders [`KnownLiveOrders`] holds before it forgets the oldest.
///
/// Venues cap an account's open orders well below this (Binance at 1,000 on spot,
/// `EXCHANGE_MAX_NUM_ORDERS`), and every order leaves the set when it ends, so a full set means
/// orders whose end was never seen, such as a terminal report that could not be converted. The
/// oldest is the likeliest to be one of those.
pub(crate) const MAX_KNOWN_LIVE_ORDERS: usize = 4_096;

/// How many recently ended client order ids [`KnownLiveOrders`] remembers, so that a late report
/// of an order as live cannot bring it back.
const RECENTLY_ENDED: NonZeroUsize = match NonZeroUsize::new(1_024) {
    Some(capacity) => capacity,
    None => panic!("a non-zero literal"),
};

/// How many order lookups run at once: in [`fetch_ended_by_key`], and per listing in
/// [`recover_ended_orders`].
pub(crate) const ORDER_LOOKUPS_IN_FLIGHT: usize = 8;

/// How many listings [`recover_ended_orders`] checks at once. With [`ORDER_LOOKUPS_IN_FLIGHT`]
/// lookups each, at most 16 requests are in flight.
const LISTINGS_IN_FLIGHT: usize = 2;

/// The time budget for one pass of [`recover_ended_orders`].
pub(crate) const ORDER_CHECK_TIMEOUT_SECS: u64 = 30;

/// The known-live orders of one client, shared by its calls and its account streams.
///
/// The account stream takes the lock for every event it observes, so nothing is logged while it
/// is held: what the set has to say waits in it until the [`KnownLiveOrdersGuard`] drops.
#[derive(Debug, Clone)]
pub(crate) struct SharedKnownLiveOrders(Arc<parking_lot::Mutex<KnownLiveOrders>>);

impl SharedKnownLiveOrders {
    /// Lock the set. What it has to say while locked is logged when the guard drops, so a guard
    /// held long defers that logging with it.
    pub(crate) fn lock(&self) -> KnownLiveOrdersGuard<'_> {
        KnownLiveOrdersGuard {
            guard: self.0.lock(),
            log: ForgottenLog::default(),
        }
    }

    /// Hold `kind` as the kind of the order `cid` while it is placed, so that a report of the
    /// order that arrives before the answer to placing it is described by it: a venue may report
    /// a new order on its account stream first. An unresolved kind is not held.
    ///
    /// Call it as the order's request is sent, and [`SendingKind::sent`] once the answer is read.
    /// When the answer leaves the outcome unknown, or the guard is dropped before it is read, as
    /// when the placing future is cancelled, the order may be live: its kind is then kept as
    /// [recalled](KnownLiveOrders::recall) until the order is seen to end.
    ///
    /// A client order id is taken to name one placement at a time: a second placement under the
    /// same id while the first is out replaces its kind, and the first one's answer releases it.
    // Only Binance margin's reports cannot always describe an order's kind.
    #[cfg_attr(not(feature = "binance"), allow(dead_code))]
    pub(crate) fn sending(&self, cid: &ClientOrderId, kind: OrderKind) -> SendingKind {
        let cid = (!kind.is_conditional_unresolved()).then(|| {
            self.lock().sending(cid, kind);
            cid.clone()
        });
        SendingKind {
            known: self.clone(),
            cid,
        }
    }

    /// How many handles share the set: a test's check that a dropped stream released its own.
    #[cfg(test)]
    pub(crate) fn holders(&self) -> usize {
        Arc::strong_count(&self.0)
    }
}

/// The lock on a [`SharedKnownLiveOrders`]. On drop it releases the lock, then logs what the set
/// kept to say.
pub(crate) struct KnownLiveOrdersGuard<'a> {
    // Fields drop in declaration order: the lock is released before `log` logs.
    guard: parking_lot::MutexGuard<'a, KnownLiveOrders>,
    log: ForgottenLog,
}

impl Deref for KnownLiveOrdersGuard<'_> {
    type Target = KnownLiveOrders;

    fn deref(&self) -> &KnownLiveOrders {
        &self.guard
    }
}

impl DerefMut for KnownLiveOrdersGuard<'_> {
    fn deref_mut(&mut self) -> &mut KnownLiveOrders {
        &mut self.guard
    }
}

impl Drop for KnownLiveOrdersGuard<'_> {
    fn drop(&mut self) {
        if !self.guard.forgotten.is_empty() {
            self.log = ForgottenLog {
                exchange: Some(self.guard.exchange),
                forgotten: std::mem::take(&mut self.guard.forgotten),
            };
        }
    }
}

/// The orders a [`KnownLiveOrdersGuard`] took from the set, logged when it is dropped, which is
/// after the lock is released.
#[derive(Default)]
struct ForgottenLog {
    exchange: Option<ExchangeId>,
    forgotten: Vec<Forgotten>,
}

impl Drop for ForgottenLog {
    fn drop(&mut self) {
        let Some(exchange) = self.exchange else {
            return;
        };
        for Forgotten { cid, first } in self.forgotten.drain(..) {
            if first {
                warn!(
                    %exchange,
                    %cid,
                    held = MAX_KNOWN_LIVE_ORDERS,
                    "Holding the most orders as live the client can, and forgetting the oldest as \
                     more arrive: if one ended while the stream was disconnected, a reconnect will \
                     not report it. Further orders forgotten are logged at debug"
                );
            } else {
                debug!(%exchange, %cid, "Forgetting the oldest order held as live");
            }
        }
    }
}

/// The kind of an order being placed, held by [`SharedKnownLiveOrders::sending`] until the answer
/// to placing it is read.
#[must_use = "dropped at once, it treats the placement's outcome as unknown"]
#[derive(Debug)]
pub(crate) struct SendingKind {
    known: SharedKnownLiveOrders,
    /// The order's id, until the kind is released. `None` when no kind was held.
    cid: Option<ClientOrderId>,
}

// Only Binance margin's reports cannot always describe an order's kind.
#[cfg_attr(not(feature = "binance"), allow(dead_code))]
impl SendingKind {
    /// Release the kind once the answer to placing the order is read, as `state`. The answer
    /// itself is recorded by [`KnownLiveOrders::placed`], before this. When it leaves the outcome
    /// unknown, the kind is kept as recalled.
    pub(crate) fn sent(mut self, state: &UnindexedOrderState) {
        if let Some(cid) = self.cid.take() {
            Self::release(&self.known, &cid, state.is_outcome_unknown());
        }
    }

    fn release(known: &SharedKnownLiveOrders, cid: &ClientOrderId, outcome_unknown: bool) {
        // The guard is dropped at the end of the statement, so the warning waits for no lock.
        let (ignored, exchange) = {
            let mut known = known.lock();
            (known.sent(cid, outcome_unknown), known.exchange)
        };
        if ignored {
            warn!(
                %exchange,
                %cid,
                limit = MAX_KNOWN_LIVE_ORDERS,
                "The kind of an order whose placement's outcome is unknown is past the limit of \
                 recalled kinds and is not kept: a report that cannot describe the order will \
                 leave its kind unresolved"
            );
        }
    }
}

impl Drop for SendingKind {
    fn drop(&mut self) {
        if let Some(cid) = self.cid.take() {
            Self::release(&self.known, &cid, true);
        }
    }
}

/// An order [`KnownLiveOrders`] forgot to make room, to log once its lock is released.
#[derive(Debug)]
struct Forgotten {
    cid: ClientOrderId,
    /// Whether it was the first of an episode, which is logged as a warning.
    first: bool,
}

/// What [`KnownLiveOrders`] keeps about one order.
#[derive(Debug)]
struct KnownLive {
    instrument: InstrumentNameExchange,
    /// The venue's id, when known, which is how a fill names its order.
    order_id: Option<OrderId>,
    quantity: Decimal,
    /// The order's kind, as first learned. An order's kind never changes, and a lookup whose
    /// answer cannot describe it reports this instead (see [`KnownLiveOrders::kind_of`]).
    kind: OrderKind,
    /// When the entry was added, as its key in [`KnownLiveOrders::by_age`].
    seq: u64,
}

/// The orders a client has seen live and not yet seen end, by client order id.
///
/// It learns of an order from the response to placing it, from a listing of open orders, and from
/// its live reports on the account stream, and drops it when it sees the order end there: a
/// terminal report, a successful cancel, or a fill that brings it to its full quantity. Whatever is
/// left after a disconnect is what a reconnect asks about.
///
/// A client order id is taken to name one order for good, as the engine takes it. One seen to end
/// is remembered for a while, so a report of it as live that arrives late, such as the response to
/// placing an order that has already filled, does not add it back.
#[derive(Debug)]
pub(crate) struct KnownLiveOrders {
    /// The venue, named in what the set logs.
    exchange: ExchangeId,
    orders: FnvHashMap<ClientOrderId, KnownLive>,
    by_order_id: FnvHashMap<(InstrumentNameExchange, OrderId), ClientOrderId>,
    by_instrument: OrdersByInstrument,
    /// Every held order by when it was added, oldest first.
    by_age: BTreeMap<u64, ClientOrderId>,
    recently_ended: LruCache<ClientOrderId, ()>,
    /// Kinds known for orders whose venue reports may not describe them, until each order is seen
    /// to end (see [`recall`](Self::recall)): ones the caller supplied, and ones kept from a
    /// placement whose outcome is unknown (see [`SharedKnownLiveOrders::sending`]).
    recalled: FnvHashMap<ClientOrderId, OrderKind>,
    /// The kinds of orders being placed, each from when its request is sent until its answer is
    /// read (see [`SharedKnownLiveOrders::sending`]).
    sending: FnvHashMap<ClientOrderId, OrderKind>,
    next_seq: u64,
    /// Whether the set is forgetting orders, so that it warns once per episode, not per order.
    forgetting: bool,
    /// The orders forgotten since the lock was taken, logged once it is released. Only a
    /// [`KnownLiveOrdersGuard`] drains it, which is why clients take the set from
    /// [`shared`](Self::shared).
    forgotten: Vec<Forgotten>,
}

impl KnownLiveOrders {
    /// A new, empty set of `exchange`'s orders. Clients take [`shared`](Self::shared): only its
    /// guard logs what the set forgets.
    fn new(exchange: ExchangeId) -> Self {
        Self {
            exchange,
            orders: FnvHashMap::default(),
            by_order_id: FnvHashMap::default(),
            by_instrument: OrdersByInstrument::default(),
            by_age: BTreeMap::new(),
            recently_ended: LruCache::new(RECENTLY_ENDED),
            recalled: FnvHashMap::default(),
            sending: FnvHashMap::default(),
            next_seq: 0,
            forgetting: false,
            forgotten: Vec::new(),
        }
    }

    /// A new, empty set of `exchange`'s orders, to share.
    pub(crate) fn shared(exchange: ExchangeId) -> SharedKnownLiveOrders {
        SharedKnownLiveOrders(Arc::new(parking_lot::Mutex::new(Self::new(exchange))))
    }

    /// Record that the order under `key`, of `quantity` and `kind`, is live as `open` says. An
    /// `open` with nothing left to fill has ended instead.
    pub(crate) fn live(
        &mut self,
        key: &UnindexedOrderKey,
        quantity: Decimal,
        kind: OrderKind,
        open: &Open,
    ) {
        let cid = &key.cid;
        if open.filled_quantity >= quantity {
            self.ended(cid);
            return;
        }
        if self.recently_ended.contains(cid) {
            return;
        }
        let order_id = open.id.assigned();
        if let Some(known) = self.orders.get_mut(cid) {
            known.quantity = quantity;
            if known.kind.is_conditional_unresolved() && !kind.is_conditional_unresolved() {
                known.kind = kind;
            }
            let moved = known.instrument != key.instrument;
            let new_id = order_id.filter(|id| known.order_id.as_ref() != Some(*id));
            // The common case, such as a partial fill: nothing to re-index. The entry under its
            // id is taken to be its own, as a venue names one order by one id.
            if !moved && new_id.is_none() {
                return;
            }
            // Re-indexed from scratch, so an entry under an id or instrument it no longer has
            // goes.
            if let Some(stale) = &known.order_id {
                self.by_order_id
                    .remove(&(known.instrument.clone(), stale.clone()));
            }
            if moved {
                self.by_instrument.remove(&known.instrument, cid);
                self.by_instrument.insert(&key.instrument, cid);
                known.instrument = key.instrument.clone();
            }
            if let Some(new_id) = new_id {
                known.order_id = Some(new_id.clone());
            }
            if let Some(order_id) = &known.order_id {
                self.by_order_id
                    .insert((key.instrument.clone(), order_id.clone()), cid.clone());
            }
            return;
        }

        if self.orders.len() >= MAX_KNOWN_LIVE_ORDERS {
            self.forget_oldest();
        } else if self.orders.len() <= MAX_KNOWN_LIVE_ORDERS * 3 / 4 {
            self.forgetting = false;
        }
        let seq = self.next_seq;
        self.next_seq += 1;
        self.by_age.insert(seq, cid.clone());
        self.by_instrument.insert(&key.instrument, cid);
        if let Some(order_id) = order_id {
            self.by_order_id
                .insert((key.instrument.clone(), order_id.clone()), cid.clone());
        }
        let kind = match self.supplied_kind(cid) {
            Some(supplied) if kind.is_conditional_unresolved() => supplied,
            _ => kind,
        };
        self.orders.insert(
            cid.clone(),
            KnownLive {
                instrument: key.instrument.clone(),
                order_id: order_id.cloned(),
                quantity,
                kind,
                seq,
            },
        );
    }

    /// Record kinds known for orders whose venue reports may not describe them: kinds the caller
    /// supplies, such as after a restart, when the client has not seen the orders placed, and the
    /// kind of an order whose placement's outcome is unknown.
    ///
    /// A recalled kind is reported for an order only when the venue's report leaves its kind
    /// unresolved and the client has not learned a resolved one itself, from placing the order or
    /// from a report that described it. An entry is dropped once its order is seen to end, and one
    /// for an order already seen to end is not kept. An unresolved kind recorded here is ignored,
    /// since it says nothing, as is anything past [`MAX_KNOWN_LIVE_ORDERS`] entries. Returns how
    /// many were ignored for that limit, for the caller to log once the lock is released.
    // Only Binance margin's reports cannot always describe an order's kind.
    #[cfg_attr(not(feature = "binance"), allow(dead_code))]
    pub(crate) fn recall(
        &mut self,
        kinds: impl IntoIterator<Item = (ClientOrderId, OrderKind)>,
    ) -> usize {
        let mut ignored = 0usize;
        for (cid, kind) in kinds {
            if kind.is_conditional_unresolved() || self.recently_ended.contains(&cid) {
                continue;
            }
            self.resolve_held(&cid, kind);
            if self.recalled.len() >= MAX_KNOWN_LIVE_ORDERS && !self.recalled.contains_key(&cid) {
                ignored += 1;
                continue;
            }
            self.recalled.insert(cid, kind);
        }
        ignored
    }

    /// Give the order `cid`, if held with an unresolved kind, the resolved `kind`.
    fn resolve_held(&mut self, cid: &ClientOrderId, kind: OrderKind) {
        if let Some(known) = self.orders.get_mut(cid)
            && known.kind.is_conditional_unresolved()
        {
            known.kind = kind;
        }
    }

    /// Hold the resolved `kind` as that of the order `cid`, about to be sent.
    fn sending(&mut self, cid: &ClientOrderId, kind: OrderKind) {
        self.resolve_held(cid, kind);
        self.sending.insert(cid.clone(), kind);
    }

    /// Release the kind held for the order `cid` while it was sent, keeping it as recalled when
    /// the outcome is unknown. Returns whether that kind was ignored for the limit.
    fn sent(&mut self, cid: &ClientOrderId, outcome_unknown: bool) -> bool {
        match self.sending.remove(cid) {
            Some(kind) if outcome_unknown => self.recall([(cid.clone(), kind)]) > 0,
            _ => false,
        }
    }

    /// Record what the response to placing the order under `key`, of `quantity` and `kind`, said:
    /// live, or already ended, which a later report of it as live then cannot undo.
    ///
    /// A response that leaves the outcome unknown
    /// ([`OrderState::is_outcome_unknown`]) records nothing: the order may be live, and a later
    /// report of it as live must still be held.
    pub(crate) fn placed(
        &mut self,
        key: &UnindexedOrderKey,
        quantity: Decimal,
        kind: OrderKind,
        state: &UnindexedOrderState,
    ) {
        match state {
            OrderState::Active(ActiveOrderState::Open(open)) => {
                self.live(key, quantity, kind, open);
            }
            OrderState::Active(_) => {}
            // Not an end: the order may be live, so a later live report must still be held.
            OrderState::Inactive(_) if state.is_outcome_unknown() => {}
            OrderState::Inactive(_) => {
                self.ended(&key.cid);
            }
        }
    }

    /// Record that a new order is about to be placed under `cid`, which names no live order: forget
    /// that an earlier order under it ended, so that this one is held as live once placed.
    ///
    /// Call it before sending the order, so that the new order's own end, which may be reported
    /// before the response to placing it, is remembered over it.
    // Only IBKR lets a client order id name a new order once its last one has ended.
    #[cfg_attr(not(feature = "ibkr"), allow(dead_code))]
    pub(crate) fn placing(&mut self, cid: &ClientOrderId) {
        self.recently_ended.pop(cid);
    }

    /// Record that the order `cid` has ended. Returns whether it was held as live.
    pub(crate) fn ended(&mut self, cid: &ClientOrderId) -> bool {
        self.recently_ended.put(cid.clone(), ());
        self.recalled.remove(cid);
        self.remove(cid)
    }

    /// Whether [`observe`](Self::observe) learns anything from an event of this kind, so a caller
    /// can skip the lock for the rest.
    pub(crate) fn observes<ExchangeKey, AssetKey, InstrumentKey>(
        kind: &AccountEventKind<ExchangeKey, AssetKey, InstrumentKey>,
    ) -> bool {
        matches!(
            kind,
            AccountEventKind::OrderSnapshot(_)
                | AccountEventKind::OrderCancelled(_)
                | AccountEventKind::Trade(_)
        )
    }

    /// Learn from an event the account stream delivers.
    ///
    /// On the stream, an [`OrderCancelled`](AccountEventKind::OrderCancelled) always reports an
    /// order leaving the book, as an error when the venue rejected it after accepting it. A cancel
    /// request that fails is answered only to its caller, and never reaches this.
    pub(crate) fn observe(&mut self, event: &UnindexedAccountEvent) {
        match &event.kind {
            AccountEventKind::OrderSnapshot(Snapshot(order)) => match &order.state {
                OrderState::Active(ActiveOrderState::Open(open)) => {
                    self.live(&order.key, order.quantity, order.kind, open);
                }
                OrderState::Active(_) => {}
                OrderState::Inactive(_) => {
                    self.ended(&order.key.cid);
                }
            },
            AccountEventKind::OrderCancelled(response) => {
                self.ended(&response.key.cid);
            }
            AccountEventKind::Trade(trade) => {
                let Some(filled) = trade.order_filled_quantity else {
                    return;
                };
                let Some(cid) = self
                    .by_order_id
                    .get(&(trade.instrument.clone(), trade.order_id.clone()))
                else {
                    return;
                };
                if self
                    .orders
                    .get(cid)
                    .is_some_and(|known| filled >= known.quantity)
                {
                    let cid = cid.clone();
                    self.ended(&cid);
                }
            }
            _ => {}
        }
    }

    /// The kind the client knows the order `cid` by: the kind it is held as live with, or else
    /// the kind it is being placed with (see [`SharedKnownLiveOrders::sending`]), or else one
    /// [recalled](Self::recall) for it.
    ///
    /// For a report whose venue answer cannot describe the order's kind, such as a Binance margin
    /// conditional order, whose REST row does not say whether it trails: the order is still
    /// reported with the kind it was placed, reported or recalled with.
    ///
    /// It is the kind first recorded for `cid`, except that an
    /// [unresolved](OrderKind::ConditionalUnresolved) kind is replaced by the first resolved one
    /// learned: a later report of the order as live does not change it otherwise. That assumes a
    /// client order id names one order for as long as it is held: an id reused for an order of
    /// another kind while held would keep the first kind. An id named again once its order has
    /// ended (see [`placing`](Self::placing)) is recorded afresh, since an order that ends is no
    /// longer held.
    // Only Binance margin's reports cannot always describe an order's kind.
    #[cfg_attr(not(feature = "binance"), allow(dead_code))]
    pub(crate) fn kind_of(&self, cid: &ClientOrderId) -> Option<OrderKind> {
        // A held order already has any kind supplied for it in place of an unresolved one:
        // `sending` and `recall` upgrade it, and `live` takes the supplied kind on adding it.
        self.orders
            .get(cid)
            .map(|known| known.kind)
            .or_else(|| self.supplied_kind(cid))
    }

    /// The kind the order `cid` is being placed with, or else one recalled for it.
    fn supplied_kind(&self, cid: &ClientOrderId) -> Option<OrderKind> {
        self.sending
            .get(cid)
            .or_else(|| self.recalled.get(cid))
            .copied()
    }

    /// Whether the order `cid` is held as live.
    #[cfg(test)]
    pub(crate) fn contains(&self, cid: &ClientOrderId) -> bool {
        self.orders.contains_key(cid)
    }

    /// Panic unless the indexes agree with the orders held.
    #[cfg(test)]
    fn assert_consistent(&self) {
        assert_eq!(self.by_age.len(), self.orders.len(), "by_age");
        for (seq, cid) in &self.by_age {
            assert_eq!(self.orders.get(cid).map(|known| known.seq), Some(*seq));
        }
        for ((instrument, order_id), cid) in &self.by_order_id {
            let known = &self.orders[cid];
            assert_eq!(
                (&known.instrument, known.order_id.as_ref()),
                (instrument, Some(order_id))
            );
        }
        let with_ids = self
            .orders
            .values()
            .filter(|known| known.order_id.is_some());
        assert_eq!(with_ids.count(), self.by_order_id.len(), "by_order_id");
        for (instrument, cids) in &self.by_instrument.0 {
            assert!(!cids.is_empty(), "by_instrument keeps no empty entry");
            for cid in cids {
                assert_eq!(&self.orders[cid].instrument, instrument);
            }
        }
        let indexed: usize = self.by_instrument.0.values().map(FnvHashSet::len).sum();
        assert_eq!(indexed, self.orders.len(), "by_instrument");
    }

    /// Every instrument with an order held as live.
    // Only Alpaca, Hyperliquid and IBKR stream every instrument.
    #[cfg_attr(
        not(any(feature = "alpaca", feature = "hyperliquid", feature = "ibkr")),
        allow(dead_code)
    )]
    pub(crate) fn instruments(&self) -> Vec<InstrumentNameExchange> {
        self.by_instrument.0.keys().cloned().collect()
    }

    /// Those of `instruments` with an order held as live.
    // Hyperliquid's stream ignores its instruments, so it checks every one held.
    #[cfg_attr(not(any(feature = "alpaca", feature = "binance")), allow(dead_code))]
    pub(crate) fn instruments_among(
        &self,
        instruments: &[InstrumentNameExchange],
    ) -> Vec<InstrumentNameExchange> {
        instruments
            .iter()
            .filter(|instrument| self.by_instrument.0.contains_key(*instrument))
            .cloned()
            .collect()
    }

    /// The keys of the orders held as live on `instrument`, under `exchange`. Venues record no
    /// strategy, so each carries [`StrategyId::unknown`].
    pub(crate) fn keys_on(
        &self,
        exchange: ExchangeId,
        instrument: &InstrumentNameExchange,
    ) -> Vec<UnindexedOrderKey> {
        self.by_instrument
            .0
            .get(instrument)
            .into_iter()
            .flatten()
            .map(|cid| UnindexedOrderKey {
                exchange,
                instrument: instrument.clone(),
                strategy: StrategyId::unknown(),
                cid: cid.clone(),
            })
            .collect()
    }

    fn remove(&mut self, cid: &ClientOrderId) -> bool {
        let Some(known) = self.orders.remove(cid) else {
            return false;
        };
        self.by_age.remove(&known.seq);
        self.by_instrument.remove(&known.instrument, cid);
        if let Some(order_id) = known.order_id {
            self.by_order_id.remove(&(known.instrument, order_id));
        }
        true
    }

    fn forget_oldest(&mut self) {
        let Some((_, cid)) = self.by_age.pop_first() else {
            return;
        };
        self.remove(&cid);
        let first = !self.forgetting;
        self.forgetting = true;
        self.forgotten.push(Forgotten { cid, first });
    }
}

/// The client order ids of the orders [`KnownLiveOrders`] holds, by instrument. An instrument is
/// present only while it has an order held, so its keys are the instruments with one.
#[derive(Debug, Default)]
struct OrdersByInstrument(FnvHashMap<InstrumentNameExchange, FnvHashSet<ClientOrderId>>);

impl OrdersByInstrument {
    fn insert(&mut self, instrument: &InstrumentNameExchange, cid: &ClientOrderId) {
        self.0
            .entry(instrument.clone())
            .or_default()
            .insert(cid.clone());
    }

    fn remove(&mut self, instrument: &InstrumentNameExchange, cid: &ClientOrderId) {
        if let Some(cids) = self.0.get_mut(instrument) {
            cids.remove(cid);
            if cids.is_empty() {
                self.0.remove(instrument);
            }
        }
    }
}

/// When a failed check of one instrument is retried.
#[derive(Debug, Default)]
struct InstrumentCheck {
    failures: u32,
    retry_at: Option<tokio::time::Instant>,
}

/// The instruments whose known-live orders a reconnect has yet to check, each with when a failed
/// check of it is retried.
///
/// A reconnect adds every instrument it recovers fills for that has an order held as live
/// ([`open`](Self::open)). An instrument is checked only once it has no fills pending
/// ([`PendingFills`]), so the fills of an order reach the stream before how it ended
/// ([`ready`](Self::ready)). A check that fails is retried after the same backoff as a Binance
/// fill gap, until it has failed
/// [`MAX_GAP_RETRIES`] retries ([`failed`](Self::failed)); a reconnect does not bring a retry
/// forward.
#[derive(Debug, Default)]
pub(crate) struct UncheckedOrders(FnvHashMap<InstrumentNameExchange, InstrumentCheck>);

impl UncheckedOrders {
    /// Add `instruments` to those to check. One already waiting keeps its retry schedule.
    pub(crate) fn open(&mut self, instruments: impl IntoIterator<Item = InstrumentNameExchange>) {
        for instrument in instruments {
            self.0.entry(instrument).or_default();
        }
    }

    /// The instruments with no fills pending whose check is due at `now`, in name order, so they
    /// are checked in a stable order.
    pub(crate) fn ready(
        &self,
        pending: &impl PendingFills,
        now: tokio::time::Instant,
    ) -> Vec<InstrumentNameExchange> {
        let mut ready: Vec<_> = self
            .0
            .iter()
            .filter(|(instrument, check)| {
                !pending.pending(instrument) && check.retry_at.is_none_or(|at| at <= now)
            })
            .map(|(instrument, _)| instrument.clone())
            .collect();
        ready.sort_unstable();
        ready
    }

    /// When an instrument with no fills pending is next due, or `None` when there is none. One
    /// with fills pending is due only once they are settled, which fill recovery's own schedule
    /// wakes for, so it never makes this a time already past that [`ready`](Self::ready) would
    /// answer with nothing.
    pub(crate) fn next_due(&self, pending: &impl PendingFills) -> Option<tokio::time::Instant> {
        self.0
            .iter()
            .filter(|(instrument, _)| !pending.pending(instrument))
            .map(|(_, check)| check.retry_at.unwrap_or_else(tokio::time::Instant::now))
            .min()
    }

    /// Record that `instrument` was checked.
    pub(crate) fn checked(&mut self, instrument: &InstrumentNameExchange) {
        self.0.remove(instrument);
    }

    /// Record that a check of `instrument` failed or did not finish at `now`: it is retried after a
    /// backoff of at least `min_delay`, or dropped once it has failed [`MAX_GAP_RETRIES`] retries.
    /// Returns which, or `None` if the instrument is not waiting.
    ///
    /// `min_delay` is the venue's own wait after refusing a request for its rate limit, or zero.
    /// A long one, such as a ban's, holds the retry back for all of it while the failure still
    /// counts toward [`MAX_GAP_RETRIES`].
    ///
    /// The base backoff is bounded by [`MAX_GAP_RETRIES`]. `min_delay` is bounded only by the
    /// client that read it from the venue, as every client in this crate bounds it, so that
    /// `now + delay` does not overflow.
    pub(crate) fn failed(
        &mut self,
        instrument: &InstrumentNameExchange,
        now: tokio::time::Instant,
        min_delay: Duration,
    ) -> Option<GapFailure> {
        let check = self.0.get_mut(instrument)?;
        check.failures += 1;
        if check.failures > MAX_GAP_RETRIES {
            self.0.remove(instrument);
            return Some(GapFailure::GivenUp);
        }
        let delay = Duration::from_secs(GAP_RETRY_BASE_SECS << (check.failures - 1)).max(min_delay);
        check.retry_at = Some(now + delay);
        Some(GapFailure::Retry(delay))
    }

    /// Hold back the retry of `instrument`, if it is waiting, until at least `until`.
    pub(crate) fn hold_until(
        &mut self,
        instrument: &InstrumentNameExchange,
        until: tokio::time::Instant,
    ) {
        if let Some(check) = self.0.get_mut(instrument) {
            check.retry_at = Some(check.retry_at.map_or(until, |retry_at| retry_at.max(until)));
        }
    }

    /// Whether `instrument` is waiting for a check.
    #[cfg(test)]
    pub(crate) fn contains(&self, instrument: &InstrumentNameExchange) -> bool {
        self.0.contains_key(instrument)
    }

    /// Whether no instrument is left to check.
    #[cfg(test)]
    pub(crate) fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
}

/// Where [`recover_ended_orders`] sends each order it finds ended: the account stream.
pub(crate) trait EventSender {
    /// Send `event` on the stream. Returns `false` once nothing more can be sent, because the
    /// consumer has gone or the stream has ended.
    fn send_event(&self, event: UnindexedAccountEvent) -> bool;
}

impl EventSender for mpsc::UnboundedSender<UnindexedAccountEvent> {
    fn send_event(&self, event: UnindexedAccountEvent) -> bool {
        self.send(event).is_ok()
    }
}

/// What looking up one order by its key found.
#[derive(Debug)]
pub(crate) enum OrderLookup {
    /// The order has ended, as this says. Boxed, as the other variants carry nothing.
    Ended(Box<UnindexedInactiveOrder>),
    /// The order is live, or in a state this version cannot read as ended.
    NotEnded,
    /// The venue does not know the order under the key's instrument.
    Unknown,
}

impl OrderLookup {
    /// How the order ended, if it has.
    pub(crate) fn ended(self) -> Option<UnindexedInactiveOrder> {
        match self {
            Self::Ended(order) => Some(*order),
            Self::NotEnded | Self::Unknown => None,
        }
    }
}

/// Answer [`OrderStatusClient::fetch_ended_orders`](crate::client::OrderStatusClient) by looking
/// each order up with `lookup`.
///
/// Keeps the trait's contract: each distinct instrument and client order id is looked up once,
/// under the first key that names it; a client order id is reported at most once, under the first
/// key that finds it; and any failed lookup fails the whole call. Up to
/// [`ORDER_LOOKUPS_IN_FLIGHT`] lookups run at once.
pub(crate) async fn fetch_ended_by_key<F, Fut>(
    orders: &[UnindexedOrderKey],
    lookup: F,
) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError>
where
    F: FnMut(UnindexedOrderKey) -> Fut,
    Fut: Future<Output = Result<OrderLookup, UnindexedClientError>>,
{
    use futures::{StreamExt as _, TryStreamExt as _};

    let mut asked = FnvHashSet::default();
    let lookups: Vec<UnindexedOrderKey> = orders
        .iter()
        .filter(|key| asked.insert((&key.instrument, &key.cid)))
        .cloned()
        .collect();
    // `buffered`, not `buffer_unordered`: the results come back in the order asked, which is what
    // makes the first key that finds an order the one it is reported under.
    let found: Vec<OrderLookup> = futures::stream::iter(lookups)
        .map(lookup)
        .buffered(ORDER_LOOKUPS_IN_FLIGHT)
        .try_collect()
        .await?;
    let mut reported = FnvHashSet::default();
    Ok(found
        .into_iter()
        .filter_map(OrderLookup::ended)
        .filter(|order| reported.insert(order.key.cid.clone()))
        .collect())
}

/// How [`recover_ended_orders`] lists the open orders it compares the held ones with.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum OpenListing {
    /// One listing per instrument, as on a venue that lists the open orders of one symbol at a
    /// time. A listing that fails charges only its own instrument.
    // Only Binance lists by one symbol.
    #[cfg_attr(not(feature = "binance"), allow(dead_code))]
    PerInstrument,
    /// One listing of every instrument due, as on a venue that lists several symbols in one
    /// request. A listing that fails charges every instrument in it. Either way, a lookup that
    /// fails charges only the instrument of the order it asked about.
    // Only Alpaca, Hyperliquid and IBKR list several symbols.
    #[cfg_attr(
        not(any(feature = "alpaca", feature = "hyperliquid", feature = "ibkr")),
        allow(dead_code)
    )]
    Batched,
}

/// Report how each order held as live ended, where it has, on every instrument due for a check:
/// the order lifecycle events missed while the stream was disconnected.
///
/// Checks only an instrument with no fills pending (see [`UncheckedOrders`]), so an order's
/// recovered fills reach the stream before how it ended. It lists the open orders with
/// `list_open`, which returns the client order ids listed on the instruments it is given, one
/// instrument or all of them at a time as `listing` says, and looks up with `lookup` each order
/// held as live that the listing no longer shows: one request per listing plus one per order that
/// ended, rather than one per order held. The orders held are read before the listings, so one
/// placed after a listing cannot be taken for one that ended.
///
/// Each order that has ended is sent as an [`AccountEventKind::OrderSnapshot`] of its inactive
/// state, under [`StrategyId::unknown`], and leaves the set. One the stream reported ending in the
/// meantime has left it already, and is not sent twice. One the venue does not know leaves the set
/// with a warning, since asking again cannot change that. One still live, or in a state this
/// version cannot read, stays.
///
/// Each lookup is settled as soon as it ends, and the instruments of each listing as soon as its
/// check ends, so dropping this part-way, as a disconnect during a retry does, loses nothing, and a
/// pass that outlasts [`ORDER_CHECK_TIMEOUT_SECS`] keeps what it finished. An instrument whose
/// listing or any lookup fails, or whose check does not finish, is retried later, asking only about
/// the orders still held; one never started stays due as it was.
///
/// A listing or lookup refused with [`ApiError::RateLimit`] stops the whole check, since every
/// request still to make would spend more against a venue already refusing. Once the refusal is
/// read no further listing or lookup starts, those already started are settled as they answer, and
/// every instrument not settled, started or not, is retried later with the usual backoff. A refusal
/// that outlasts the backoff can therefore give up on an instrument never asked about.
/// Any other failure charges only its own instrument, and the check goes on.
///
/// Every instrument the check retries later, for whatever reason, waits at least the longest wait
/// the venue's refusals in this check asked for, since a request sent sooner would be refused too.
/// A longer wait read after an instrument was charged still holds it back, unless the check is
/// dropped first.
pub(crate) async fn recover_ended_orders<L, LFut, Q, QFut>(
    exchange: ExchangeId,
    known: &SharedKnownLiveOrders,
    unchecked: &mut UncheckedOrders,
    pending: &impl PendingFills,
    tx: &impl EventSender,
    listing: OpenListing,
    list_open: L,
    lookup: Q,
) where
    L: Fn(Vec<InstrumentNameExchange>) -> LFut,
    LFut: Future<Output = Result<FnvHashSet<ClientOrderId>, UnindexedClientError>>,
    Q: Fn(UnindexedOrderKey) -> QFut,
    QFut: Future<Output = Result<OrderLookup, UnindexedClientError>>,
{
    use futures::StreamExt as _;

    let ready = unchecked.ready(pending, tokio::time::Instant::now());
    if ready.is_empty() {
        return;
    }
    let mut held: Vec<(InstrumentNameExchange, Vec<UnindexedOrderKey>)> = {
        let known = known.lock();
        ready
            .into_iter()
            .map(|instrument| {
                let keys = known.keys_on(exchange, &instrument);
                (instrument, keys)
            })
            .collect()
    };
    held.retain(|(instrument, keys)| {
        if keys.is_empty() {
            unchecked.checked(instrument);
        }
        !keys.is_empty()
    });
    if held.is_empty() {
        return;
    }
    info!(
        %exchange,
        instruments = held.len(),
        orders = held.iter().map(|(_, keys)| keys.len()).sum::<usize>(),
        "Checking how the orders held as live ended while disconnected"
    );

    // One check per listing.
    let checks: Vec<Vec<(InstrumentNameExchange, Vec<UnindexedOrderKey>)>> = match listing {
        OpenListing::PerInstrument => held.into_iter().map(|one| vec![one]).collect(),
        OpenListing::Batched => vec![held],
    };
    // Which checks have been settled, and how many have started: the stream starts them in order,
    // so those from `started` on were never read. The stream owns what it reads, so no future it
    // builds borrows from a closure argument, which `tokio::spawn` cannot prove `Send`.
    let instruments: Vec<Vec<InstrumentNameExchange>> = checks
        .iter()
        .map(|check| {
            check
                .iter()
                .map(|(instrument, _)| instrument.clone())
                .collect()
        })
        .collect();
    let mut settled = vec![false; checks.len()];
    let started = AtomicUsize::new(0);
    // Set by the first request the venue refuses for its rate limit, and read before any other
    // starts.
    let throttled = Throttled::default();
    let (list_open, lookup, throttled) = (&list_open, &lookup, &throttled);
    // Instruments this check charged with a failure, held back once it ends until the longest wait
    // the venue asked for, which a request still in flight when one was charged can raise.
    let mut charged: Vec<&InstrumentNameExchange> = Vec::new();
    let recovery = async {
        let mut reported = 0u32;
        // Instruments retried for the venue's rate limit, logged once rather than each.
        let mut deferred = 0usize;
        let mut runs =
            futures::stream::iter(checks.into_iter().enumerate().map(|(index, check)| {
                started.store(index + 1, Ordering::Relaxed);
                async move {
                    if throttled.stopped() {
                        return (
                            index,
                            Err(UnindexedClientError::Api(ApiError::rate_limited(None))),
                        );
                    }
                    let (listed_on, keys): (Vec<_>, Vec<_>) = check.into_iter().unzip();
                    let listed = match list_open(listed_on).await {
                        Ok(listed) => listed,
                        Err(e) => {
                            throttled.note(&e);
                            return (index, Err(e));
                        }
                    };
                    let unlisted: Vec<UnindexedOrderKey> = keys
                        .into_iter()
                        .flatten()
                        .filter(|key| !listed.contains(&key.cid))
                        .collect();
                    // How many lookups each instrument has still to settle, so a check stopped
                    // for the rate limit charges those it never started.
                    let mut unsettled: FnvHashMap<InstrumentNameExchange, usize> =
                        FnvHashMap::default();
                    for key in &unlisted {
                        *unsettled.entry(key.instrument.clone()).or_default() += 1;
                    }
                    // Each lookup is settled as it ends, so one that fails, or a pass that times
                    // out, loses none of those already answered. Once the venue has refused one
                    // for its rate limit, no further lookup is taken from the source, and those
                    // already started are settled as they answer.
                    let mut lookups = futures::stream::iter(unlisted)
                        .take_while(|_| std::future::ready(!throttled.stopped()))
                        .map(|key| async move {
                            let found = lookup(key.clone()).await;
                            (key, found)
                        })
                        .buffer_unordered(ORDER_LOOKUPS_IN_FLIGHT);
                    let mut checked = CheckedListing::default();
                    while let Some((key, found)) = lookups.next().await {
                        if let Some(left) = unsettled.get_mut(&key.instrument) {
                            *left -= 1;
                        }
                        match found {
                            Ok(found) => match settle(exchange, known, key, found, tx) {
                                Some(sent) => checked.sent += sent,
                                None => {
                                    checked.consumer_gone = true;
                                    return (index, Ok(checked));
                                }
                            },
                            Err(e) => {
                                throttled.note(&e);
                                checked
                                    .failed
                                    .entry(key.instrument)
                                    .or_insert_with(|| failure_reason(&e));
                            }
                        }
                    }
                    for (instrument, left) in unsettled {
                        if left > 0 {
                            checked
                                .failed
                                .entry(instrument)
                                .or_insert_with(|| RATE_LIMITED.to_owned());
                        }
                    }
                    (index, Ok(checked))
                }
            }))
            .buffer_unordered(LISTINGS_IN_FLIGHT);
        while let Some((index, checked)) = runs.next().await {
            settled[index] = true;
            match checked {
                Ok(checked) => {
                    if checked.consumer_gone {
                        debug!(%exchange, "Order check: consumer dropped during recovery");
                        return;
                    }
                    reported += checked.sent;
                    for instrument in &instruments[index] {
                        match checked.failed.get(instrument) {
                            Some(reason) => {
                                deferred += usize::from(order_check_failed(
                                    exchange,
                                    unchecked,
                                    instrument,
                                    reason,
                                    throttled.wait(),
                                ));
                                charged.push(instrument);
                            }
                            None => unchecked.checked(instrument),
                        }
                    }
                }
                Err(e) => {
                    let reason = failure_reason(&e);
                    for instrument in &instruments[index] {
                        deferred += usize::from(order_check_failed(
                            exchange,
                            unchecked,
                            instrument,
                            &reason,
                            throttled.wait(),
                        ));
                        charged.push(instrument);
                    }
                }
            }
        }
        if throttled.stopped() {
            warn!(
                %exchange,
                reported,
                deferred,
                venue_wait_ms = u64::try_from(throttled.wait().as_millis()).unwrap_or(u64::MAX),
                "Order check stopped: the venue is limiting requests, so the instruments not \
                 checked are retried later"
            );
        } else {
            info!(%exchange, reported, "Order check complete");
        }
    };
    if tokio::time::timeout(Duration::from_secs(ORDER_CHECK_TIMEOUT_SECS), recovery)
        .await
        .is_err()
    {
        let started = started.load(Ordering::Relaxed);
        for (check, _) in instruments[..started]
            .iter()
            .zip(&settled)
            .filter(|(_, settled)| !**settled)
        {
            for instrument in check {
                order_check_failed(
                    exchange,
                    unchecked,
                    instrument,
                    "order check timed out",
                    throttled.wait(),
                );
                charged.push(instrument);
            }
        }
    }
    let venue_wait = throttled.wait();
    if !venue_wait.is_zero() {
        let until = tokio::time::Instant::now() + venue_wait;
        for instrument in charged {
            unchecked.hold_until(instrument, until);
        }
    }
}

/// Why an instrument is retried when the venue refused a request for its rate limit, whether the
/// refused request was its own or not.
const RATE_LIMITED: &str = "the venue is limiting requests";

/// Whether `error` is the venue refusing a request for its rate limit, which stops an order check.
fn is_rate_limit(error: &UnindexedClientError) -> bool {
    matches!(error, UnindexedClientError::Api(ApiError::RateLimit { .. }))
}

/// Whether the venue has refused one of an order check's requests for its rate limit, and the
/// longest wait its refusals asked for.
#[derive(Debug, Default)]
struct Throttled {
    stopped: AtomicBool,
    /// The longest `retry_after` seen, in milliseconds.
    wait_ms: AtomicU64,
}

impl Throttled {
    /// Stop the check if `error` is the venue refusing a request for its rate limit, keeping the
    /// longest wait asked for.
    fn note(&self, error: &UnindexedClientError) {
        if let UnindexedClientError::Api(ApiError::RateLimit { retry_after, .. }) = error {
            if let Some(wait) = retry_after {
                let wait_ms = u64::try_from(wait.as_millis()).unwrap_or(u64::MAX);
                self.wait_ms.fetch_max(wait_ms, Ordering::Relaxed);
            }
            self.stopped.store(true, Ordering::Relaxed);
        }
    }

    fn stopped(&self) -> bool {
        self.stopped.load(Ordering::Relaxed)
    }

    /// The longest wait the venue's refusals asked for: zero when none gave one.
    fn wait(&self) -> Duration {
        Duration::from_millis(self.wait_ms.load(Ordering::Relaxed))
    }
}

/// What an instrument whose check failed with `error` is logged as retried for.
fn failure_reason(error: &UnindexedClientError) -> String {
    if is_rate_limit(error) {
        RATE_LIMITED.to_owned()
    } else {
        error.to_string()
    }
}

/// What one listing's lookups came to, once each has been settled.
#[derive(Debug, Default)]
struct CheckedListing {
    /// How many orders that ended were sent.
    sent: u32,
    /// The instruments a lookup failed on, each with the first failure's reason.
    failed: FnvHashMap<InstrumentNameExchange, String>,
    /// Whether the consumer has gone, so the check stopped.
    consumer_gone: bool,
}

/// Apply one lookup to the set and send the order if it ended. Returns how many were sent (none or
/// one), or `None` if the consumer has gone.
///
/// The set's lock is held across the send, as the stream holds it across its own, so an order the
/// stream reports ending and this check reach the consumer in the order they were decided, and an
/// order is reported once.
fn settle(
    exchange: ExchangeId,
    known: &SharedKnownLiveOrders,
    key: UnindexedOrderKey,
    found: OrderLookup,
    tx: &impl EventSender,
) -> Option<u32> {
    let mut known = known.lock();
    match found {
        OrderLookup::Ended(order) => {
            if !known.ended(&order.key.cid) {
                return Some(0);
            }
            let event = UnindexedAccountEvent::new(
                exchange,
                AccountEventKind::OrderSnapshot(Snapshot::new(
                    (*order).map_state(OrderState::Inactive),
                )),
            );
            tx.send_event(event).then_some(1)
        }
        OrderLookup::Unknown => {
            let held = known.ended(&key.cid);
            drop(known);
            if held {
                warn!(
                    %exchange,
                    instrument = %key.instrument,
                    cid = %key.cid,
                    "The venue does not know an order held as live; a reconnect no longer asks \
                     about it"
                );
            }
            Some(0)
        }
        OrderLookup::NotEnded => Some(0),
    }
}

/// Record that the check of how the orders held as live on `instrument` ended did not finish,
/// because of `reason`, and log what follows: a retry, or giving the check up.
///
/// Returns whether it is a retry for the venue's rate limit, which is logged at debug: the check
/// stopped for it warns once for all of them.
///
/// The retry waits at least `venue_wait`, the longest wait the venue's refusals in this check asked
/// for so far, whatever the reason: a request sent sooner would be refused too.
fn order_check_failed(
    exchange: ExchangeId,
    unchecked: &mut UncheckedOrders,
    instrument: &InstrumentNameExchange,
    reason: &str,
    venue_wait: Duration,
) -> bool {
    let rate_limited = reason == RATE_LIMITED;
    match unchecked.failed(instrument, tokio::time::Instant::now(), venue_wait) {
        Some(GapFailure::Retry(delay)) => {
            if rate_limited {
                debug!(
                    %exchange,
                    %instrument,
                    retry_in_secs = delay.as_secs(),
                    reason,
                    "Could not check how the orders held as live ended, retrying later"
                );
            } else {
                warn!(
                    %exchange,
                    %instrument,
                    retry_in_secs = delay.as_secs(),
                    reason,
                    "Could not check how the orders held as live ended, retrying later"
                );
            }
            rate_limited
        }
        Some(GapFailure::GivenUp) => {
            error!(
                %exchange,
                %instrument,
                retries = MAX_GAP_RETRIES,
                reason,
                "Gave up checking how the orders held as live ended: an order that ended \
                 while disconnected stays live in engine state until the next reconnect checks it; \
                 reconcile with fetch_open_orders"
            );
            false
        }
        None => false,
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
pub(crate) mod tests {
    use super::*;
    use crate::{
        error::{ConnectivityError, OrderError},
        order::{
            Order, OrderKey, OrderKind, TimeInForce,
            id::VenueOrderId,
            request::UnindexedOrderResponseCancel,
            state::{Cancelled, Expired, Filled, InactiveOrderState, OpenInFlight},
        },
        trade::{AssetFees, Trade, TradeId},
    };
    use chrono::Utc;
    use rust_decimal_macros::dec;
    use rustrade_instrument::{Side, asset::name::AssetNameExchange};

    /// Fills pending on `instrument` alone.
    fn pending_on(instrument: &InstrumentNameExchange) -> FnvHashSet<InstrumentNameExchange> {
        FnvHashSet::from_iter([instrument.clone()])
    }

    impl PendingFills for FnvHashSet<InstrumentNameExchange> {
        fn pending(&self, instrument: &InstrumentNameExchange) -> bool {
            self.contains(instrument)
        }
    }

    fn key(instrument: &str, cid: &str) -> UnindexedOrderKey {
        OrderKey::new(
            ExchangeId::BinanceSpot,
            InstrumentNameExchange::new(instrument),
            StrategyId::unknown(),
            ClientOrderId::new(cid),
        )
    }

    fn open(order_id: &str, filled: Decimal) -> Open {
        Open::new(
            VenueOrderId::Assigned(OrderId::new(order_id)),
            Utc::now(),
            filled,
        )
    }

    fn event(
        kind: AccountEventKind<ExchangeId, AssetNameExchange, InstrumentNameExchange>,
    ) -> UnindexedAccountEvent {
        UnindexedAccountEvent::new(ExchangeId::BinanceSpot, kind)
    }

    fn snapshot(
        key: UnindexedOrderKey,
        state: OrderState<AssetNameExchange, InstrumentNameExchange>,
    ) -> UnindexedAccountEvent {
        event(AccountEventKind::OrderSnapshot(Snapshot(Order {
            key,
            side: Side::Buy,
            price: Some(dec!(100)),
            quantity: dec!(2),
            kind: OrderKind::Limit,
            time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
            state,
        })))
    }

    fn fill(order_id: &str, order_filled: Option<Decimal>) -> UnindexedAccountEvent {
        event(AccountEventKind::Trade(Trade::new(
            TradeId::new("t"),
            OrderId::new(order_id),
            InstrumentNameExchange::new("BTCUSDT"),
            StrategyId::unknown(),
            Utc::now(),
            Side::Buy,
            dec!(100),
            dec!(1),
            order_filled,
            AssetFees::new(AssetNameExchange::new("usdt"), Decimal::ZERO, None),
        )))
    }

    /// The set keeps the kind an order was first learned with, from a placement or a report, and
    /// forgets it with the order.
    #[test]
    fn the_set_keeps_the_kind_an_order_was_first_learned_with() {
        let trailing_stop = OrderKind::TrailingStop {
            offset: dec!(1.5),
            offset_type: crate::order::TrailingOffsetType::Percentage,
            activation_price: None,
        };
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let order = key("BTCUSDT", "trailing");
        known.placed(
            &order,
            dec!(2),
            trailing_stop,
            &OrderState::active(open("7", Decimal::ZERO)),
        );
        assert_eq!(known.kind_of(&order.cid), Some(trailing_stop));

        // A later report, here describing the trail in basis points, does not replace it.
        known.live(
            &order,
            dec!(2),
            OrderKind::TrailingStop {
                offset: dec!(150),
                offset_type: crate::order::TrailingOffsetType::BasisPoints,
                activation_price: None,
            },
            &open("7", dec!(1)),
        );
        assert_eq!(known.kind_of(&order.cid), Some(trailing_stop));

        known.ended(&order.cid);
        assert_eq!(known.kind_of(&order.cid), None);
    }

    fn unresolved() -> OrderKind {
        OrderKind::ConditionalUnresolved {
            family: crate::order::ConditionalFamily::StopLimit,
            trigger_price: Some(dec!(96)),
        }
    }

    fn trailing_fixed_limit() -> OrderKind {
        OrderKind::TrailingStopFixedLimit {
            offset: dec!(25),
            offset_type: crate::order::TrailingOffsetType::BasisPoints,
            activation_price: Some(dec!(96)),
        }
    }

    /// An order first learned with an unresolved kind takes the first resolved kind reported for
    /// it, and a resolved kind is never replaced by an unresolved one.
    #[test]
    fn an_unresolved_kind_is_replaced_by_a_resolved_one_and_never_the_reverse() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceMargin);
        let order = key("BTCUSDT", "stop");
        known.live(&order, dec!(2), unresolved(), &open("7", Decimal::ZERO));
        assert_eq!(known.kind_of(&order.cid), Some(unresolved()));

        known.live(
            &order,
            dec!(2),
            trailing_fixed_limit(),
            &open("7", Decimal::ZERO),
        );
        assert_eq!(known.kind_of(&order.cid), Some(trailing_fixed_limit()));

        known.live(&order, dec!(2), unresolved(), &open("7", dec!(1)));
        assert_eq!(known.kind_of(&order.cid), Some(trailing_fixed_limit()));
    }

    /// A recalled kind is the order's known kind until the client learns a resolved one itself:
    /// it stands in for an unresolved report, held or not, but not for a resolved one, and it is
    /// dropped with the order's end.
    #[test]
    fn a_recalled_kind_stands_in_for_an_unresolved_one_until_the_order_ends() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceMargin);
        let held = key("BTCUSDT", "held");
        let new = key("BTCUSDT", "new");
        let placed = key("BTCUSDT", "placed");
        let fixed = OrderKind::StopLimit {
            trigger_price: dec!(96),
        };
        known.live(&held, dec!(2), unresolved(), &open("1", Decimal::ZERO));
        known.placed(
            &placed,
            dec!(2),
            fixed,
            &OrderState::active(open("3", Decimal::ZERO)),
        );
        known.recall([
            (held.cid.clone(), trailing_fixed_limit()),
            (new.cid.clone(), trailing_fixed_limit()),
            (placed.cid.clone(), trailing_fixed_limit()),
        ]);

        // A held unresolved order takes it at once; one not yet seen has it before it is seen,
        // and keeps it once seen unresolved.
        assert_eq!(known.kind_of(&held.cid), Some(trailing_fixed_limit()));
        assert_eq!(known.kind_of(&new.cid), Some(trailing_fixed_limit()));
        known.live(&new, dec!(2), unresolved(), &open("2", Decimal::ZERO));
        assert_eq!(known.kind_of(&new.cid), Some(trailing_fixed_limit()));
        // The client's own resolved kind wins, whether learned before the recall or after.
        assert_eq!(known.kind_of(&placed.cid), Some(fixed));
        let placed_later = key("BTCUSDT", "placed later");
        known.recall([(placed_later.cid.clone(), trailing_fixed_limit())]);
        known.placed(
            &placed_later,
            dec!(2),
            fixed,
            &OrderState::active(open("4", Decimal::ZERO)),
        );
        assert_eq!(known.kind_of(&placed_later.cid), Some(fixed));

        known.ended(&new.cid);
        assert_eq!(known.kind_of(&new.cid), None);
        assert!(!known.recalled.contains_key(&new.cid));
    }

    /// A recall is ignored for an order already seen to end and for an unresolved kind, and past
    /// the limit.
    #[test]
    fn a_recall_ignores_ended_orders_unresolved_kinds_and_entries_past_the_limit() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceMargin);
        let ended = ClientOrderId::new("ended");
        known.ended(&ended);
        let ignored = known.recall([
            (ended.clone(), trailing_fixed_limit()),
            (ClientOrderId::new("unresolved"), unresolved()),
        ]);
        assert!(known.recalled.is_empty());
        assert_eq!(ignored, 0, "only what the limit leaves out is counted");

        let ignored = known.recall((0..MAX_KNOWN_LIVE_ORDERS + 2).map(|n| {
            (
                ClientOrderId::new(format!("cid-{n}")),
                trailing_fixed_limit(),
            )
        }));
        assert_eq!(ignored, 2);
        assert_eq!(known.recalled.len(), MAX_KNOWN_LIVE_ORDERS);
        assert!(
            !known
                .recalled
                .contains_key(&ClientOrderId::new(format!("cid-{MAX_KNOWN_LIVE_ORDERS}")))
        );
        // A kind already held is updated, at the limit too.
        known.recall([(ClientOrderId::new("cid-0"), unresolved())]);
        let fixed = OrderKind::StopLimit {
            trigger_price: dec!(1),
        };
        assert_eq!(known.recall([(ClientOrderId::new("cid-0"), fixed)]), 0);
        assert_eq!(known.kind_of(&ClientOrderId::new("cid-0")), Some(fixed));
    }

    /// The kind an order is being placed with describes a report of it that arrives before the
    /// answer to placing it, and the order stays held with that kind once the answer is read.
    #[test]
    fn a_kind_being_placed_describes_a_report_that_arrives_before_the_answer() {
        let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
        let order = key("BTCUSDT", "racing");

        let sending = known.sending(&order.cid, trailing_fixed_limit());
        assert_eq!(
            known.lock().kind_of(&order.cid),
            Some(trailing_fixed_limit())
        );
        known
            .lock()
            .live(&order, dec!(2), unresolved(), &open("7", Decimal::ZERO));
        assert_eq!(
            known.lock().kind_of(&order.cid),
            Some(trailing_fixed_limit())
        );

        let answer = OrderState::active(open("7", Decimal::ZERO));
        known
            .lock()
            .placed(&order, dec!(2), trailing_fixed_limit(), &answer);
        sending.sent(&answer);
        let held = known.lock();
        assert!(held.sending.is_empty() && held.recalled.is_empty());
        assert_eq!(held.kind_of(&order.cid), Some(trailing_fixed_limit()));
    }

    /// An order already held with an unresolved kind, such as under an id placed again, takes the
    /// kind it is being placed with.
    #[test]
    fn a_kind_being_placed_resolves_a_held_unresolved_one() {
        let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
        let order = key("BTCUSDT", "held");
        known
            .lock()
            .live(&order, dec!(2), unresolved(), &open("7", Decimal::ZERO));

        let sending = known.sending(&order.cid, trailing_fixed_limit());
        assert_eq!(
            known.lock().kind_of(&order.cid),
            Some(trailing_fixed_limit())
        );
        sending.sent(&OrderState::active(open("7", Decimal::ZERO)));
        assert_eq!(
            known.lock().kind_of(&order.cid),
            Some(trailing_fixed_limit())
        );
    }

    /// Once the answer is read, the kind an order was placed with is kept only when the order may
    /// be live: the outcome is unknown, or the guard was dropped unread. An unresolved kind is
    /// never held.
    #[test]
    fn a_placed_kind_is_kept_after_the_answer_only_when_the_outcome_is_unknown() {
        let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
        let cid = |name: &str| ClientOrderId::new(name);

        known
            .sending(&cid("rejected"), trailing_fixed_limit())
            .sent(&OrderState::inactive(OrderError::Rejected(
                ApiError::OrderRejected("refused".to_string()),
            )));
        assert_eq!(known.lock().kind_of(&cid("rejected")), None);

        known
            .sending(&cid("unknown"), trailing_fixed_limit())
            .sent(&OrderState::inactive(OrderError::Connectivity(
                ConnectivityError::Timeout,
            )));
        assert_eq!(
            known.lock().kind_of(&cid("unknown")),
            Some(trailing_fixed_limit())
        );
        known.lock().ended(&cid("unknown"));
        assert_eq!(known.lock().kind_of(&cid("unknown")), None);

        drop(known.sending(&cid("cancelled"), trailing_fixed_limit()));
        assert_eq!(
            known.lock().kind_of(&cid("cancelled")),
            Some(trailing_fixed_limit())
        );

        let sending = known.sending(&cid("unresolved"), unresolved());
        assert_eq!(known.lock().kind_of(&cid("unresolved")), None);
        drop(sending);
        assert_eq!(known.lock().kind_of(&cid("unresolved")), None);
        assert!(known.lock().sending.is_empty());
    }

    /// A kind kept for an unknown outcome past the limit of recalled kinds is warned of, once the
    /// lock is released.
    #[test]
    fn a_placed_kind_past_the_limit_is_warned_of_with_the_lock_released() {
        let known = KnownLiveOrders::shared(ExchangeId::BinanceMargin);
        known.lock().recall((0..MAX_KNOWN_LIVE_ORDERS).map(|n| {
            (
                ClientOrderId::new(format!("cid-{n}")),
                trailing_fixed_limit(),
            )
        }));
        let past = ClientOrderId::new("past");

        let seen = logged_with_lock_free(&known, || {
            drop(known.sending(&past, trailing_fixed_limit()));
        });
        assert_eq!(seen, vec![(tracing::Level::WARN, true)]);
        assert_eq!(known.lock().kind_of(&past), None);

        let seen = logged_with_lock_free(&known, || {
            known
                .sending(&ClientOrderId::new("cid-0"), trailing_fixed_limit())
                .sent(&OrderState::inactive(OrderError::Connectivity(
                    ConnectivityError::Timeout,
                )));
        });
        assert_eq!(seen, vec![], "a kind already recalled is within the limit");
    }

    /// An id whose order ended is not held again by a late report of it as live, but a new order
    /// placed under it is.
    #[test]
    fn a_new_order_under_an_ended_id_is_held_once_placing_it() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let reused = key("BTCUSDT", "reused");
        known.ended(&reused.cid);
        known.live(
            &reused,
            dec!(2),
            OrderKind::Limit,
            &open("1", Decimal::ZERO),
        );
        assert!(!known.contains(&reused.cid));

        known.placing(&reused.cid);
        known.live(
            &reused,
            dec!(2),
            OrderKind::Limit,
            &open("2", Decimal::ZERO),
        );
        assert!(known.contains(&reused.cid));
        known.assert_consistent();
    }

    #[test]
    fn the_stream_adds_a_live_order_and_drops_it_at_each_kind_of_end() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let cancelled = key("BTCUSDT", "cancelled");
        let rejected = key("BTCUSDT", "rejected");
        let inactive = key("BTCUSDT", "inactive");
        for (key, id) in [(&cancelled, "1"), (&rejected, "2"), (&inactive, "3")] {
            known.observe(&snapshot(
                key.clone(),
                OrderState::active(open(id, Decimal::ZERO)),
            ));
            assert!(known.contains(&key.cid));
        }

        known.observe(&event(AccountEventKind::OrderCancelled(
            UnindexedOrderResponseCancel {
                key: cancelled.clone(),
                state: Ok(Cancelled::new(
                    OrderId::new("1"),
                    Utc::now(),
                    Some(Decimal::ZERO),
                )),
            },
        )));
        known.observe(&event(AccountEventKind::OrderCancelled(
            UnindexedOrderResponseCancel {
                key: rejected.clone(),
                state: Err(OrderError::Rejected(ApiError::OrderRejected("no".into()))),
            },
        )));
        known.observe(&snapshot(
            inactive.clone(),
            OrderState::Inactive(InactiveOrderState::Cancelled(Cancelled::new(
                OrderId::new("3"),
                Utc::now(),
                Some(Decimal::ZERO),
            ))),
        ));

        assert!(
            known
                .instruments_among(&[InstrumentNameExchange::new("BTCUSDT")])
                .is_empty(),
            "every order ended"
        );
    }

    #[test]
    fn a_fill_drops_its_order_only_once_it_reports_the_whole_quantity() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let order = key("BTCUSDT", "a");
        known.live(&order, dec!(2), OrderKind::Limit, &open("7", Decimal::ZERO));

        known.observe(&fill("7", None));
        assert!(known.contains(&order.cid), "no cumulative says nothing");
        known.observe(&fill("7", Some(dec!(1))));
        assert!(known.contains(&order.cid), "half filled is still live");
        known.observe(&fill("8", Some(dec!(2))));
        assert!(known.contains(&order.cid), "another order's fill");
        known.assert_consistent();
        known.observe(&fill("7", Some(dec!(2))));
        assert!(!known.contains(&order.cid), "filled");
        known.assert_consistent();
    }

    #[test]
    fn a_report_without_the_venue_id_keeps_the_one_known() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let order = key("BTCUSDT", "a");
        known.live(&order, dec!(2), OrderKind::Limit, &open("7", Decimal::ZERO));
        known.live(
            &order,
            dec!(2),
            OrderKind::Limit,
            &Open::new(VenueOrderId::ClientAssigned, Utc::now(), dec!(1)),
        );

        known.assert_consistent();
        known.observe(&fill("7", Some(dec!(2))));
        assert!(!known.contains(&order.cid), "still found by its venue id");
        known.assert_consistent();
    }

    #[test]
    fn an_order_with_nothing_left_or_already_ended_is_not_added() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let filled = key("BTCUSDT", "filled");
        known.live(&filled, dec!(2), OrderKind::Limit, &open("1", dec!(2)));
        assert!(!known.contains(&filled.cid));

        let late = key("BTCUSDT", "late");
        known.live(&late, dec!(2), OrderKind::Limit, &open("2", Decimal::ZERO));
        assert!(known.ended(&late.cid));
        known.live(&late, dec!(2), OrderKind::Limit, &open("2", Decimal::ZERO));
        assert!(
            !known.contains(&late.cid),
            "a late live report does not bring it back"
        );
        assert!(!known.ended(&late.cid), "and it is no longer held");
    }

    #[test]
    fn a_placement_holds_an_open_order_and_settles_an_ended_one() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let open_order = key("BTCUSDT", "open");
        known.placed(
            &open_order,
            dec!(2),
            OrderKind::Limit,
            &OrderState::active(open("1", Decimal::ZERO)),
        );
        assert!(known.contains(&open_order.cid));

        let in_flight = key("BTCUSDT", "in-flight");
        known.placed(
            &in_flight,
            dec!(2),
            OrderKind::Limit,
            &OrderState::active(OpenInFlight::new(Utc::now())),
        );
        assert!(
            !known.contains(&in_flight.cid),
            "not yet known to the venue"
        );

        let ended = [
            (
                "filled",
                OrderState::fully_filled(Filled::new(OrderId::new("2"), Utc::now(), dec!(2), None)),
            ),
            (
                "expired",
                OrderState::inactive(Expired::new(OrderId::new("3"), Utc::now(), Some(dec!(1)))),
            ),
            (
                "failed",
                OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected("no".into()))),
            ),
        ];
        for (cid, state) in ended {
            let order = key("BTCUSDT", cid);
            known.placed(&order, dec!(2), OrderKind::Limit, &state);
            known.live(&order, dec!(2), OrderKind::Limit, &open("9", Decimal::ZERO));
            assert!(
                !known.contains(&order.cid),
                "{cid}: a late live report does not bring it back"
            );
        }
        known.assert_consistent();
    }

    /// An open answered with an unknown outcome may be live, so it is not remembered as ended: a
    /// later report of it as live holds it.
    #[test]
    fn a_placement_with_an_unknown_outcome_is_held_once_reported_live() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let order = key("BTCUSDT", "unknown");
        known.placed(
            &order,
            dec!(2),
            OrderKind::Limit,
            &OrderState::inactive(OrderError::Connectivity(ConnectivityError::Timeout)),
        );
        assert!(!known.contains(&order.cid), "not yet known to be live");
        known.live(&order, dec!(2), OrderKind::Limit, &open("9", Decimal::ZERO));
        assert!(known.contains(&order.cid), "held once reported live");
        known.assert_consistent();
    }

    /// A report that changes neither the venue id nor the instrument leaves the indexes as they
    /// were, and one that changes either re-indexes the order under its new key only.
    #[test]
    fn a_live_report_re_indexes_only_what_changed() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let order = key("BTCUSDT", "a");
        known.live(&order, dec!(2), OrderKind::Limit, &open("7", Decimal::ZERO));
        known.live(&order, dec!(2), OrderKind::Limit, &open("7", dec!(1)));
        known.assert_consistent();

        known.live(&order, dec!(2), OrderKind::Limit, &open("8", dec!(1)));
        known.assert_consistent();
        known.observe(&fill("7", Some(dec!(2))));
        assert!(known.contains(&order.cid), "no longer found by its old id");

        let no_id = Open::new(VenueOrderId::ClientAssigned, Utc::now(), dec!(1));
        known.live(&order, dec!(2), OrderKind::Limit, &no_id);
        known.assert_consistent();

        // Moved, with no id reported: the id known moves with it.
        let moved = key("ETHUSDT", "a");
        known.live(&moved, dec!(2), OrderKind::Limit, &no_id);
        known.assert_consistent();
        assert_eq!(
            known.instruments(),
            [InstrumentNameExchange::new("ETHUSDT")]
        );

        let back = key("BTCUSDT", "a");
        known.live(&back, dec!(2), OrderKind::Limit, &open("8", dec!(1)));
        known.assert_consistent();
        assert_eq!(
            known.instruments(),
            [InstrumentNameExchange::new("BTCUSDT")]
        );
        known.observe(&fill("8", Some(dec!(2))));
        assert!(
            !known.contains(&order.cid),
            "found by its id on its instrument"
        );
    }

    /// Run `f`, returning the level of every event it logged on this thread and whether `known`'s
    /// lock was free when it was logged.
    pub(crate) fn logged_with_lock_free(
        known: &SharedKnownLiveOrders,
        f: impl FnOnce(),
    ) -> Vec<(tracing::Level, bool)> {
        use std::sync::Mutex;
        use tracing_subscriber::layer::SubscriberExt;

        struct LockFree {
            known: SharedKnownLiveOrders,
            seen: Arc<Mutex<Vec<(tracing::Level, bool)>>>,
        }

        impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for LockFree {
            fn on_event(
                &self,
                event: &tracing::Event<'_>,
                _: tracing_subscriber::layer::Context<'_, S>,
            ) {
                let free = self.known.0.try_lock().is_some();
                self.seen
                    .lock()
                    .unwrap()
                    .push((*event.metadata().level(), free));
            }
        }

        let seen = Arc::new(Mutex::new(Vec::new()));
        let layer = LockFree {
            known: known.clone(),
            seen: Arc::clone(&seen),
        };
        // Thread-local, so other tests running in parallel are not captured.
        tracing::subscriber::with_default(tracing_subscriber::registry().with(layer), f);
        seen.lock().unwrap().clone()
    }

    /// The set is locked by the account stream for every event, so what it logs on forgetting an
    /// order waits until the lock is released.
    #[test]
    fn forgetting_is_logged_with_the_lock_released() {
        let known = KnownLiveOrders::shared(ExchangeId::BinanceSpot);

        let seen = logged_with_lock_free(&known, || {
            let mut held = known.lock();
            for n in 0..MAX_KNOWN_LIVE_ORDERS + 2 {
                held.live(
                    &key("BTCUSDT", &n.to_string()),
                    dec!(1),
                    OrderKind::Limit,
                    &open(&n.to_string(), Decimal::ZERO),
                );
            }
            assert_eq!(held.forgotten.len(), 2, "kept while held");
        });

        assert_eq!(
            seen,
            [(tracing::Level::WARN, true), (tracing::Level::DEBUG, true)]
        );
        assert!(known.lock().forgotten.is_empty());
        assert!(!known.lock().contains(&ClientOrderId::new("1")));
    }

    #[test]
    fn an_order_the_venue_does_not_know_is_logged_with_the_lock_released() {
        let known = KnownLiveOrders::shared(ExchangeId::BinanceSpot);
        let order = key("BTCUSDT", "a");
        known
            .lock()
            .live(&order, dec!(2), OrderKind::Limit, &open("7", Decimal::ZERO));
        let (tx, _rx) = mpsc::unbounded_channel();

        let seen = logged_with_lock_free(&known, || {
            let sent = settle(
                ExchangeId::BinanceSpot,
                &known,
                order.clone(),
                OrderLookup::Unknown,
                &tx,
            );
            assert_eq!(sent, Some(0));
        });

        assert_eq!(seen, [(tracing::Level::WARN, true)]);
        assert!(!known.lock().contains(&order.cid));
    }

    #[test]
    fn a_full_set_forgets_its_oldest_order() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        // The oldest is alone on its instrument, so forgetting it stops holding that instrument.
        for n in 0..=MAX_KNOWN_LIVE_ORDERS {
            let instrument = if n == 0 { "ETHUSDT" } else { "BTCUSDT" };
            known.live(
                &key(instrument, &n.to_string()),
                dec!(1),
                OrderKind::Limit,
                &open(&n.to_string(), Decimal::ZERO),
            );
        }
        assert!(!known.contains(&ClientOrderId::new("0")));
        assert!(known.contains(&ClientOrderId::new("1")));
        assert_eq!(
            known.instruments(),
            [InstrumentNameExchange::new("BTCUSDT")]
        );
        // An order that ends frees its place, so the next one forgets nothing.
        assert!(known.ended(&ClientOrderId::new("2")));
        known.live(
            &key("BTCUSDT", "new"),
            dec!(1),
            OrderKind::Limit,
            &open("new", Decimal::ZERO),
        );
        assert!(known.contains(&ClientOrderId::new("1")));
        known.live(
            &key("BTCUSDT", "newer"),
            dec!(1),
            OrderKind::Limit,
            &open("newer", Decimal::ZERO),
        );
        assert!(
            !known.contains(&ClientOrderId::new("1")),
            "then the oldest again"
        );
        known.assert_consistent();
        assert!(known.contains(&ClientOrderId::new(MAX_KNOWN_LIVE_ORDERS.to_string())));
        assert_eq!(
            known
                .keys_on(
                    ExchangeId::BinanceSpot,
                    &InstrumentNameExchange::new("BTCUSDT")
                )
                .len(),
            MAX_KNOWN_LIVE_ORDERS
        );
    }

    #[test]
    fn keys_on_names_only_that_instrument() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        known.live(
            &key("BTCUSDT", "a"),
            dec!(1),
            OrderKind::Limit,
            &open("1", Decimal::ZERO),
        );
        known.live(
            &key("ETHUSDT", "b"),
            dec!(1),
            OrderKind::Limit,
            &open("2", Decimal::ZERO),
        );

        assert_eq!(
            known.keys_on(
                ExchangeId::BinanceSpot,
                &InstrumentNameExchange::new("ETHUSDT")
            ),
            [key("ETHUSDT", "b")]
        );
        let (btc, eth, sol) = (
            InstrumentNameExchange::new("BTCUSDT"),
            InstrumentNameExchange::new("ETHUSDT"),
            InstrumentNameExchange::new("SOLUSDT"),
        );
        assert_eq!(
            known.instruments_among(&[sol, eth.clone()]),
            [eth],
            "only those asked about that hold an order"
        );
        assert_eq!(known.instruments_among(std::slice::from_ref(&btc)), [btc]);
        known.assert_consistent();
    }

    #[test]
    fn an_instrument_stops_being_held_with_its_last_order() {
        let mut known = KnownLiveOrders::new(ExchangeId::BinanceSpot);
        let (btc, eth) = (
            InstrumentNameExchange::new("BTCUSDT"),
            InstrumentNameExchange::new("ETHUSDT"),
        );
        known.live(
            &key("BTCUSDT", "a"),
            dec!(1),
            OrderKind::Limit,
            &open("1", Decimal::ZERO),
        );
        known.live(
            &key("BTCUSDT", "b"),
            dec!(1),
            OrderKind::Limit,
            &open("2", Decimal::ZERO),
        );

        known.ended(&ClientOrderId::new("a"));
        assert_eq!(
            known.instruments(),
            std::slice::from_ref(&btc),
            "b still holds it"
        );
        known.assert_consistent();

        known.live(
            &key("ETHUSDT", "b"),
            dec!(1),
            OrderKind::Limit,
            &open("2", Decimal::ZERO),
        );
        assert_eq!(
            known.instruments(),
            std::slice::from_ref(&eth),
            "b moved instrument"
        );
        assert!(known.keys_on(ExchangeId::BinanceSpot, &btc).is_empty());
        assert_eq!(
            known.keys_on(ExchangeId::BinanceSpot, &eth),
            [key("ETHUSDT", "b")]
        );
        known.assert_consistent();

        known.ended(&ClientOrderId::new("b"));
        assert!(known.instruments().is_empty());
        assert!(known.instruments_among(&[btc, eth]).is_empty());
        known.assert_consistent();
    }

    #[test]
    fn an_instrument_with_a_fill_gap_waits_for_it() {
        let btc = InstrumentNameExchange::new("BTCUSDT");
        let eth = InstrumentNameExchange::new("ETHUSDT");
        let pending = pending_on(&btc);
        let mut unchecked = UncheckedOrders::default();
        unchecked.open([btc.clone(), eth.clone()]);

        assert_eq!(
            unchecked.ready(&pending, tokio::time::Instant::now()),
            std::slice::from_ref(&eth)
        );
        unchecked.checked(&eth);
        assert!(unchecked.contains(&btc), "BTCUSDT still waits");
        assert_eq!(
            unchecked.next_due(&pending),
            None,
            "on its gap, not on a timer"
        );

        let pending = NoPendingFills;
        assert_eq!(
            unchecked.ready(&pending, tokio::time::Instant::now()),
            std::slice::from_ref(&btc)
        );
        assert!(unchecked.next_due(&pending).is_some(), "due now");
        unchecked.checked(&btc);
        assert!(unchecked.is_empty());
    }

    /// A retry that falls due while the instrument is back on a fill gap must not wake the loop for
    /// a check `ready` would refuse, or it spins until the gap settles.
    #[test]
    fn a_due_retry_on_an_instrument_back_on_a_gap_does_not_wake_the_loop() {
        let btc = InstrumentNameExchange::new("BTCUSDT");
        let mut unchecked = UncheckedOrders::default();
        unchecked.open([btc.clone()]);
        let start = tokio::time::Instant::now();
        unchecked.failed(&btc, start, Duration::ZERO);
        let pending = pending_on(&btc);
        let later = start + Duration::from_secs(GAP_RETRY_BASE_SECS * 2);

        assert!(unchecked.ready(&pending, later).is_empty());
        assert_eq!(unchecked.next_due(&pending), None);
        assert_eq!(
            unchecked.ready(&NoPendingFills, later),
            std::slice::from_ref(&btc),
            "due again once the gap is settled"
        );
    }

    #[test]
    fn instruments_are_retried_on_their_own_schedules() {
        let (btc, eth) = (
            InstrumentNameExchange::new("BTCUSDT"),
            InstrumentNameExchange::new("ETHUSDT"),
        );
        let mut unchecked = UncheckedOrders::default();
        unchecked.open([btc.clone(), eth.clone()]);
        let now = tokio::time::Instant::now();
        unchecked.failed(&btc, now, Duration::ZERO);
        unchecked.failed(&btc, now, Duration::ZERO);
        unchecked.failed(&eth, now, Duration::ZERO);
        let pending = NoPendingFills;

        let base = Duration::from_secs(GAP_RETRY_BASE_SECS);
        assert_eq!(
            unchecked.ready(&pending, now + base),
            std::slice::from_ref(&eth)
        );
        assert_eq!(unchecked.next_due(&pending), Some(now + base));
        unchecked.open([btc.clone()]);
        assert!(
            unchecked.ready(&pending, now + base).len() == 1,
            "a reconnect does not bring BTCUSDT's retry forward"
        );
        assert_eq!(unchecked.ready(&pending, now + base * 2).len(), 2);
    }

    #[test]
    fn a_failed_check_backs_off_and_is_given_up_after_its_retries() {
        let mut unchecked = UncheckedOrders::default();
        unchecked.open([InstrumentNameExchange::new("BTCUSDT")]);
        let btc = InstrumentNameExchange::new("BTCUSDT");
        let pending = NoPendingFills;
        let now = tokio::time::Instant::now();

        for retry in 0..MAX_GAP_RETRIES {
            let delay = Duration::from_secs(GAP_RETRY_BASE_SECS << retry);
            assert_eq!(
                unchecked.failed(&btc, now, Duration::ZERO),
                Some(GapFailure::Retry(delay))
            );
            assert_eq!(unchecked.next_due(&pending), Some(now + delay));
            assert!(unchecked.ready(&pending, now).is_empty(), "waits");
            assert_eq!(unchecked.ready(&pending, now + delay).len(), 1);
        }
        assert_eq!(
            unchecked.failed(&btc, now, Duration::ZERO),
            Some(GapFailure::GivenUp)
        );
        assert!(unchecked.is_empty());
        assert_eq!(unchecked.next_due(&pending), None);
        assert_eq!(
            unchecked.failed(&btc, now, Duration::ZERO),
            None,
            "no longer waiting"
        );
    }

    fn cancelled_order(key: &UnindexedOrderKey) -> UnindexedInactiveOrder {
        Order {
            key: key.clone(),
            side: Side::Buy,
            price: Some(dec!(100)),
            quantity: dec!(1),
            kind: OrderKind::Limit,
            time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
            state: InactiveOrderState::Cancelled(Cancelled::new(
                OrderId::new("1"),
                Utc::now(),
                Some(Decimal::ZERO),
            )),
        }
    }

    #[tokio::test]
    async fn each_order_is_looked_up_once_and_reported_under_the_first_key_that_finds_it() {
        let wrong = key("ETHUSDT", "a");
        let right = key("BTCUSDT", "a");
        let mut repeated = key("BTCUSDT", "a");
        repeated.strategy = StrategyId::new("other");
        let live = key("BTCUSDT", "live");
        let asked = std::sync::Mutex::new(Vec::new());

        let ended = fetch_ended_by_key(
            &[wrong.clone(), right.clone(), repeated, live.clone(), live],
            |key| {
                asked.lock().unwrap().push(key.clone());
                async move {
                    Ok(if key.instrument.name() == "BTCUSDT" && key.cid.0 == "a" {
                        OrderLookup::Ended(Box::new(cancelled_order(&key)))
                    } else if key.cid.0 == "a" {
                        OrderLookup::Unknown
                    } else {
                        OrderLookup::NotEnded
                    })
                }
            },
        )
        .await
        .unwrap();

        let reported: Vec<_> = ended.iter().map(|order| &order.key).collect();
        assert_eq!(reported, [&right], "under the key that found it, once");
        assert_eq!(
            *asked.lock().unwrap(),
            [wrong, right, key("BTCUSDT", "live")],
            "a repeated instrument and cid is not asked again"
        );
    }

    #[tokio::test]
    async fn one_failed_lookup_fails_the_call() {
        let result = fetch_ended_by_key(
            &[key("BTCUSDT", "a"), key("BTCUSDT", "b")],
            |key| async move {
                if key.cid.0 == "b" {
                    Err(UnindexedClientError::Api(ApiError::rate_limited(None)))
                } else {
                    Ok(OrderLookup::Ended(Box::new(cancelled_order(&key))))
                }
            },
        )
        .await;

        assert!(matches!(
            result,
            Err(UnindexedClientError::Api(ApiError::RateLimit { .. }))
        ));
    }

    #[tokio::test]
    async fn nothing_asked_is_nothing_looked_up() {
        let ended = fetch_ended_by_key(&[], |_| async {
            panic!("nothing to look up");
        })
        .await
        .unwrap();
        assert!(ended.is_empty());
    }

    /// One order held on each of `instruments`, all due for a check.
    fn held_on(instruments: &[&str]) -> (SharedKnownLiveOrders, UncheckedOrders) {
        let known = KnownLiveOrders::shared(ExchangeId::BinanceSpot);
        let mut unchecked = UncheckedOrders::default();
        for (n, instrument) in instruments.iter().enumerate() {
            known.lock().live(
                &key(instrument, instrument),
                dec!(1),
                OrderKind::Limit,
                &open(&n.to_string(), Decimal::ZERO),
            );
            unchecked.open([InstrumentNameExchange::new(*instrument)]);
        }
        (known, unchecked)
    }

    /// Run a check whose listing never answers on an instrument named `SLOW…`, and lists nothing
    /// open on the others, whose orders have all ended.
    async fn check_with_slow_instruments(
        known: &SharedKnownLiveOrders,
        unchecked: &mut UncheckedOrders,
    ) -> Vec<UnindexedAccountEvent> {
        let (tx, mut rx) = mpsc::unbounded_channel();
        recover_ended_orders(
            ExchangeId::BinanceSpot,
            known,
            unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::PerInstrument,
            |instruments: Vec<InstrumentNameExchange>| async move {
                if instruments
                    .iter()
                    .any(|instrument| instrument.name().starts_with("SLOW"))
                {
                    std::future::pending::<()>().await;
                }
                Ok(FnvHashSet::default())
            },
            |key: UnindexedOrderKey| async move {
                Ok(OrderLookup::Ended(Box::new(cancelled_order(&key))))
            },
        )
        .await;
        std::iter::from_fn(|| rx.try_recv().ok()).collect()
    }

    /// An instrument whose check timed out waits as long as the venue asked in another
    /// instrument's refusal.
    #[tokio::test(start_paused = true)]
    async fn a_timed_out_instrument_waits_for_the_venues_wait() {
        let (known, mut unchecked) = held_on(&["AAA", "SLOW"]);
        let (tx, _rx) = mpsc::unbounded_channel();
        let long = Duration::from_secs(GAP_RETRY_BASE_SECS * 10);

        recover_ended_orders(
            ExchangeId::BinanceSpot,
            &known,
            &mut unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::PerInstrument,
            |instruments: Vec<InstrumentNameExchange>| async move {
                if instruments[0].name().as_str() == "SLOW" {
                    std::future::pending::<()>().await;
                }
                // SLOW's listing is in flight before AAA's is refused.
                tokio::task::yield_now().await;
                Err(UnindexedClientError::Api(ApiError::rate_limited(Some(
                    long,
                ))))
            },
            |_: UnindexedOrderKey| async { panic!("nothing listed, nothing looked up") },
        )
        .await;

        let timed_out = tokio::time::Instant::now();
        assert!(
            unchecked
                .ready(&NoPendingFills, timed_out + long - Duration::from_secs(1))
                .is_empty(),
            "retried before the venue's wait"
        );
        assert_eq!(
            unchecked.ready(&NoPendingFills, timed_out + long * 2),
            [
                InstrumentNameExchange::new("AAA"),
                InstrumentNameExchange::new("SLOW")
            ]
        );
    }

    /// Two instruments are checked at a time, in name order. When the pass times out, the two
    /// started and unfinished have failed once, and the one never started stays due as it was.
    #[tokio::test(start_paused = true)]
    async fn a_timed_out_pass_charges_only_the_instruments_it_started() {
        let (known, mut unchecked) = held_on(&["SLOWA", "SLOWB", "XFAST"]);
        let start = tokio::time::Instant::now();

        let sent = check_with_slow_instruments(&known, &mut unchecked).await;

        assert!(sent.is_empty(), "XFAST never started: {sent:?}");
        let pending = NoPendingFills;
        let after = start + Duration::from_secs(ORDER_CHECK_TIMEOUT_SECS);
        assert_eq!(
            unchecked.ready(&pending, after),
            [InstrumentNameExchange::new("XFAST")],
            "only the one never started is due at once"
        );
        let retry = after + Duration::from_secs(GAP_RETRY_BASE_SECS);
        assert_eq!(unchecked.ready(&pending, retry).len(), 3);
        for instrument in ["SLOWA", "SLOWB"] {
            assert_eq!(
                unchecked.failed(
                    &InstrumentNameExchange::new(instrument),
                    retry,
                    Duration::ZERO
                ),
                Some(GapFailure::Retry(Duration::from_secs(
                    GAP_RETRY_BASE_SECS * 2
                ))),
                "{instrument} has failed once already"
            );
        }
        assert!(known.lock().contains(&ClientOrderId::new("XFAST")));
    }

    /// An instrument settled before the pass times out keeps its result.
    #[tokio::test(start_paused = true)]
    async fn a_timed_out_pass_keeps_what_it_finished() {
        let (known, mut unchecked) = held_on(&["AFAST", "SLOWB", "SLOWC"]);

        let sent = check_with_slow_instruments(&known, &mut unchecked).await;

        let [event] = sent.as_slice() else {
            panic!("AFAST's order: {sent:?}");
        };
        let AccountEventKind::OrderSnapshot(Snapshot(order)) = &event.kind else {
            panic!("an order snapshot: {event:?}");
        };
        assert_eq!(order.key.cid, ClientOrderId::new("AFAST"));
        assert!(!unchecked.contains(&InstrumentNameExchange::new("AFAST")));
        assert!(!known.lock().contains(&ClientOrderId::new("AFAST")));
        assert!(
            unchecked
                .ready(&NoPendingFills, tokio::time::Instant::now())
                .is_empty(),
            "SLOWB and SLOWC both started, so both wait for a retry"
        );
        known.lock().assert_consistent();
    }

    /// A batch's lookup that fails charges only its own instrument, and every lookup that was
    /// answered, on that instrument too, is kept.
    #[tokio::test]
    async fn a_failed_lookup_in_a_batch_charges_only_its_instrument() {
        let known = KnownLiveOrders::shared(ExchangeId::BinanceSpot);
        for (instrument, cid, id) in [
            ("BTCUSDT", "btc-ended", "1"),
            ("BTCUSDT", "btc-failing", "2"),
            ("ETHUSDT", "eth-ended", "3"),
        ] {
            known.lock().live(
                &key(instrument, cid),
                dec!(1),
                OrderKind::Limit,
                &open(id, Decimal::ZERO),
            );
        }
        let mut unchecked = UncheckedOrders::default();
        unchecked.open(known.lock().instruments());
        let (tx, mut rx) = mpsc::unbounded_channel();
        let listings = AtomicUsize::new(0);

        recover_ended_orders(
            ExchangeId::BinanceSpot,
            &known,
            &mut unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::Batched,
            |instruments: Vec<InstrumentNameExchange>| {
                listings.fetch_add(1, Ordering::Relaxed);
                assert_eq!(instruments.len(), 2, "one listing of both");
                async { Ok(FnvHashSet::default()) }
            },
            |key: UnindexedOrderKey| async move {
                if key.cid.0 == "btc-failing" {
                    Err(UnindexedClientError::Connectivity(
                        ConnectivityError::Timeout,
                    ))
                } else {
                    Ok(OrderLookup::Ended(Box::new(cancelled_order(&key))))
                }
            },
        )
        .await;

        assert_eq!(listings.load(Ordering::Relaxed), 1);
        let mut sent: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
            .map(|event| match event.kind {
                AccountEventKind::OrderSnapshot(Snapshot(order)) => order.key.cid,
                other => panic!("an order snapshot: {other:?}"),
            })
            .collect();
        sent.sort();
        assert_eq!(
            sent,
            [
                ClientOrderId::new("btc-ended"),
                ClientOrderId::new("eth-ended")
            ]
        );
        let btc = InstrumentNameExchange::new("BTCUSDT");
        assert!(!unchecked.contains(&InstrumentNameExchange::new("ETHUSDT")));
        assert!(unchecked.contains(&btc), "BTCUSDT waits for a retry");
        assert_eq!(
            unchecked.failed(&btc, tokio::time::Instant::now(), Duration::ZERO),
            Some(GapFailure::Retry(Duration::from_secs(
                GAP_RETRY_BASE_SECS * 2
            ))),
            "and has failed once already"
        );
        let known = known.lock();
        assert!(known.contains(&ClientOrderId::new("btc-failing")));
        assert!(!known.contains(&ClientOrderId::new("btc-ended")));
        known.assert_consistent();
    }

    /// Due for one retry, having failed once.
    fn failed_once(unchecked: &mut UncheckedOrders, instrument: &str) -> bool {
        unchecked.failed(
            &InstrumentNameExchange::new(instrument),
            tokio::time::Instant::now(),
            Duration::ZERO,
        ) == Some(GapFailure::Retry(Duration::from_secs(
            GAP_RETRY_BASE_SECS * 2,
        )))
    }

    /// A lookup refused for the rate limit stops the whole check: no other listing or lookup
    /// starts, and every instrument is retried.
    #[tokio::test]
    async fn a_rate_limited_lookup_stops_the_whole_check() {
        let (known, mut unchecked) = held_on(&["AAA", "BBB", "CCC"]);
        let (tx, mut rx) = mpsc::unbounded_channel();
        let (listings, lookups) = (AtomicUsize::new(0), AtomicUsize::new(0));

        recover_ended_orders(
            ExchangeId::BinanceSpot,
            &known,
            &mut unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::PerInstrument,
            |_| {
                listings.fetch_add(1, Ordering::Relaxed);
                async { Ok(FnvHashSet::default()) }
            },
            |_| {
                lookups.fetch_add(1, Ordering::Relaxed);
                async { Err(UnindexedClientError::Api(ApiError::rate_limited(None))) }
            },
        )
        .await;

        assert_eq!(lookups.load(Ordering::Relaxed), 1);
        assert_eq!(listings.load(Ordering::Relaxed), 1);
        assert!(rx.try_recv().is_err(), "nothing reported");
        for instrument in ["AAA", "BBB", "CCC"] {
            assert!(failed_once(&mut unchecked, instrument), "{instrument}");
        }
        known.lock().assert_consistent();
    }

    /// A rate-limited instrument is retried no sooner than the venue's refusal asked, even when
    /// that is longer than the backoff.
    #[tokio::test]
    async fn a_rate_limit_hint_longer_than_the_backoff_holds_the_retry_back() {
        let (known, mut unchecked) = held_on(&["AAA", "BBB"]);
        let (tx, _rx) = mpsc::unbounded_channel();
        let wait = Duration::from_secs(GAP_RETRY_BASE_SECS * 10);
        let started = tokio::time::Instant::now();

        recover_ended_orders(
            ExchangeId::BinanceSpot,
            &known,
            &mut unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::PerInstrument,
            |_| async { Ok(FnvHashSet::default()) },
            move |_| async move {
                Err(UnindexedClientError::Api(ApiError::rate_limited(Some(
                    wait,
                ))))
            },
        )
        .await;

        let due = unchecked.next_due(&NoPendingFills).unwrap();
        assert!(due >= started + wait, "retried before the venue's wait");
        assert!(
            unchecked
                .ready(&NoPendingFills, started + wait - Duration::from_secs(1))
                .is_empty()
        );
        assert_eq!(
            unchecked.ready(&NoPendingFills, tokio::time::Instant::now() + wait),
            [
                InstrumentNameExchange::new("AAA"),
                InstrumentNameExchange::new("BBB")
            ]
        );
    }

    /// A longer wait the venue asks for after an instrument was charged still holds that
    /// instrument back, whether a shorter wait or another failure charged it, as it does the
    /// instrument never asked about.
    #[tokio::test]
    async fn a_longer_wait_read_later_holds_back_every_instrument_charged() {
        let long = Duration::from_secs(GAP_RETRY_BASE_SECS * 10);
        let first_failures = [
            UnindexedClientError::Api(ApiError::rate_limited(Some(Duration::from_secs(1)))),
            UnindexedClientError::Connectivity(ConnectivityError::Timeout),
        ];
        for first in first_failures {
            let (known, mut unchecked) = held_on(&["AAA", "BBB", "CCC"]);
            let (tx, _rx) = mpsc::unbounded_channel();
            let started = tokio::time::Instant::now();

            recover_ended_orders(
                ExchangeId::BinanceSpot,
                &known,
                &mut unchecked,
                &NoPendingFills,
                &tx,
                OpenListing::PerInstrument,
                |instruments: Vec<InstrumentNameExchange>| {
                    let first = first.clone();
                    async move {
                        // AAA's and BBB's listings are both in flight before either fails, and
                        // AAA's failure is read first.
                        if instruments[0].name().as_str() == "AAA" {
                            tokio::task::yield_now().await;
                            return Err(first);
                        }
                        for _ in 0..3 {
                            tokio::task::yield_now().await;
                        }
                        Err(UnindexedClientError::Api(ApiError::rate_limited(Some(
                            long,
                        ))))
                    }
                },
                |_: UnindexedOrderKey| async { panic!("nothing listed, nothing looked up") },
            )
            .await;

            assert!(
                unchecked
                    .ready(&NoPendingFills, started + long - Duration::from_secs(1))
                    .is_empty(),
                "{first:?}: retried before the longest wait"
            );
            assert_eq!(
                unchecked
                    .ready(&NoPendingFills, tokio::time::Instant::now() + long)
                    .len(),
                3,
                "{first:?}"
            );
        }
    }

    #[test]
    fn a_retry_waits_the_longer_of_the_backoff_and_the_minimum() {
        let (_, mut unchecked) = held_on(&["AAA"]);
        let aaa = InstrumentNameExchange::new("AAA");
        let now = tokio::time::Instant::now();
        let backoff = Duration::from_secs(GAP_RETRY_BASE_SECS);

        assert_eq!(
            unchecked.failed(&aaa, now, backoff * 3),
            Some(GapFailure::Retry(backoff * 3))
        );
        assert_eq!(
            unchecked.failed(&aaa, now, Duration::from_secs(1)),
            Some(GapFailure::Retry(backoff * 2)),
            "a shorter minimum leaves the backoff"
        );
    }

    #[test]
    fn a_check_stopped_by_the_rate_limit_keeps_the_longest_wait() {
        let throttled = Throttled::default();
        assert!(!throttled.stopped());

        throttled.note(&UnindexedClientError::Connectivity(
            ConnectivityError::Timeout,
        ));
        assert!(!throttled.stopped(), "only a rate limit stops the check");

        throttled.note(&UnindexedClientError::Api(ApiError::rate_limited(None)));
        assert!(throttled.stopped());
        assert_eq!(throttled.wait(), Duration::ZERO);

        for secs in [300, 900, 60] {
            throttled.note(&UnindexedClientError::Api(ApiError::rate_limited(Some(
                Duration::from_secs(secs),
            ))));
        }
        assert_eq!(throttled.wait(), Duration::from_secs(900));
    }

    /// In a batch, a lookup refused for the rate limit charges every instrument it refused.
    #[tokio::test]
    async fn a_rate_limited_batch_charges_the_lookups_it_did_not_settle() {
        let (known, mut unchecked) = held_on(&["AAA", "BBB"]);
        let (tx, mut rx) = mpsc::unbounded_channel();
        let lookups = AtomicUsize::new(0);

        recover_ended_orders(
            ExchangeId::BinanceSpot,
            &known,
            &mut unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::Batched,
            |_| async { Ok(FnvHashSet::default()) },
            |_| {
                lookups.fetch_add(1, Ordering::Relaxed);
                async { Err(UnindexedClientError::Api(ApiError::rate_limited(None))) }
            },
        )
        .await;

        assert!(lookups.load(Ordering::Relaxed) <= 2);
        assert!(rx.try_recv().is_err(), "nothing reported");
        for instrument in ["AAA", "BBB"] {
            assert!(failed_once(&mut unchecked, instrument), "{instrument}");
        }
        assert_eq!(known.lock().instruments().len(), 2, "both still held");
    }

    /// Lookups that answer later, as over a network: of the first batch started, one is refused
    /// for the rate limit and the rest are settled as they answer, and none is started after. The
    /// instruments never asked about are retried with the refused one.
    #[tokio::test]
    async fn a_rate_limited_lookup_settles_those_in_flight_and_starts_no_more() {
        let names: Vec<String> = (0..ORDER_LOOKUPS_IN_FLIGHT + 2)
            .map(|n| format!("I{n}"))
            .collect();
        let (known, mut unchecked) = held_on(&names.iter().map(String::as_str).collect::<Vec<_>>());
        let (tx, mut rx) = mpsc::unbounded_channel();
        let lookups = AtomicUsize::new(0);

        recover_ended_orders(
            ExchangeId::BinanceSpot,
            &known,
            &mut unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::Batched,
            |_| async { Ok(FnvHashSet::default()) },
            |key: UnindexedOrderKey| {
                let first = lookups.fetch_add(1, Ordering::Relaxed) == 0;
                async move {
                    // Every lookup answers later, so all those started are sent before the
                    // refusal is read, and the refusal first, so no answer frees a place for
                    // another lookup before it.
                    tokio::task::yield_now().await;
                    if first {
                        return Err(UnindexedClientError::Api(ApiError::rate_limited(None)));
                    }
                    tokio::task::yield_now().await;
                    Ok(OrderLookup::Ended(Box::new(cancelled_order(&key))))
                }
            },
        )
        .await;

        assert_eq!(lookups.load(Ordering::Relaxed), ORDER_LOOKUPS_IN_FLIGHT);
        let reported = std::iter::from_fn(|| rx.try_recv().ok()).count();
        assert_eq!(
            reported,
            ORDER_LOOKUPS_IN_FLIGHT - 1,
            "those in flight are settled"
        );
        let retried = names
            .iter()
            .filter(|name| failed_once(&mut unchecked, name))
            .count();
        assert_eq!(retried, 3, "the refused one and the two never asked about");
        assert_eq!(known.lock().instruments().len(), 3);
        known.lock().assert_consistent();
    }

    /// A listing refused for the rate limit stops the check too.
    #[tokio::test]
    async fn a_rate_limited_listing_stops_the_whole_check() {
        let (known, mut unchecked) = held_on(&["AAA", "BBB", "CCC"]);
        let (tx, _rx) = mpsc::unbounded_channel();
        let listings = AtomicUsize::new(0);

        recover_ended_orders(
            ExchangeId::BinanceSpot,
            &known,
            &mut unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::PerInstrument,
            |_| {
                listings.fetch_add(1, Ordering::Relaxed);
                async { Err(UnindexedClientError::Api(ApiError::rate_limited(None))) }
            },
            |_: UnindexedOrderKey| async { panic!("nothing listed, nothing looked up") },
        )
        .await;

        assert_eq!(listings.load(Ordering::Relaxed), 1);
        for instrument in ["AAA", "BBB", "CCC"] {
            assert!(failed_once(&mut unchecked, instrument), "{instrument}");
        }
    }

    /// A batch whose listing never answers charges every instrument in it once when the pass
    /// times out.
    #[tokio::test(start_paused = true)]
    async fn a_timed_out_batch_charges_every_instrument_in_it() {
        let (known, mut unchecked) = held_on(&["AAA", "BBB"]);
        let (tx, mut rx) = mpsc::unbounded_channel();

        recover_ended_orders(
            ExchangeId::BinanceSpot,
            &known,
            &mut unchecked,
            &NoPendingFills,
            &tx,
            OpenListing::Batched,
            |_: Vec<InstrumentNameExchange>| async {
                std::future::pending::<()>().await;
                Ok(FnvHashSet::default())
            },
            |_: UnindexedOrderKey| async { Ok(OrderLookup::NotEnded) },
        )
        .await;

        assert!(rx.try_recv().is_err(), "nothing reported");
        let now = tokio::time::Instant::now();
        assert!(
            unchecked.ready(&NoPendingFills, now).is_empty(),
            "both wait"
        );
        for instrument in ["AAA", "BBB"] {
            assert_eq!(
                unchecked.failed(
                    &InstrumentNameExchange::new(instrument),
                    now,
                    Duration::ZERO
                ),
                Some(GapFailure::Retry(Duration::from_secs(
                    GAP_RETRY_BASE_SECS * 2
                ))),
                "{instrument} has failed once"
            );
        }
    }
}