perf(station): moonproto 98420c6d and a markets_version-keyed coin cache

crates/moon-core/src/market/mod.rs

@@ -25,14 +25,14 @@ use crate::session::CoreId;
 pub use candles::{CandleViewCfg, ChartCandle};
 pub use screener::{ScreenerRow, SharedScreenerRow};
 pub use source::{
-    ArbQuote, ArbVenue, CORE_ARCHIVE_RETENTION_MS, CandleReadParams, ChartHistoryBuffers,
-    ChartHistoryCursor, ChartHistoryRead, CoinTag, DetectSnapshot, LatestPriceError,
-    LiqSpanReadout, LiveTicks, MarketContextReadout, MarketDataSource, MarketFiguresReadout,
-    MarketLabel, MarketLimits, MarketQuantityUnit, MarketRevisions, MarketTickerReadout,
-    MarketWindowsReadout, MaxOrder, MaxOrderSource, OrderSizeRules, PathPrefixRows,
-    PriceLineUpdate, PriceProfileRow, ProfileWindow, ReplayAddress, ReplayAddressError,
-    ReplayPriceLinesRead, SIDE_BUCKET_MS, SideVolumeBucket, VolumeAt, VolumeSpan,
-    VolumeSpanReadout, WindowFigures, pick_market_for_coin, pick_market_for_identity,
+    ArbQuote, ArbVenue, CORE_ARCHIVE_RETENTION_MS, CandleReadParams, CatalogLookup,
+    ChartHistoryBuffers, ChartHistoryCursor, ChartHistoryRead, CoinTag, DetectSnapshot,
+    LatestPriceError, LiqSpanReadout, LiveTicks, MarketContextReadout, MarketDataSource,
+    MarketFiguresReadout, MarketLabel, MarketLimits, MarketQuantityUnit, MarketRevisions,
+    MarketTickerReadout, MarketWindowsReadout, MaxOrder, MaxOrderSource, OrderSizeRules,
+    PathPrefixRows, PriceLineUpdate, PriceProfileRow, ProfileWindow, ReplayAddress,
+    ReplayAddressError, ReplayPriceLinesRead, SIDE_BUCKET_MS, SideVolumeBucket, VolumeAt,
+    VolumeSpan, VolumeSpanReadout, WindowFigures, pick_market_for_coin, pick_market_for_identity,
     side_slots_of_ticks,
 };
 

crates/moon-core/src/market/source/catalog_lookup.rs

@@ -0,0 +1,156 @@
+//! One provider snapshot answering a run of market lookups.
+//!
+//! [`MarketDataSource::resolve_market`] read the catalog twice — the name search and the labels of
+//! what it found — each from whatever snapshot was current at that moment. A [`CatalogLookup`]
+//! holds one snapshot for both and carries that snapshot's `MarketsState::markets_version`, so a
+//! cache keyed on the version can never pair an answer from a new list with an old version.
+
+use moonproto::state::MarketsState;
+
+use super::MarketDataSource;
+use crate::feed::SharedMoonClient;
+use crate::market::MarketLabel;
+use crate::session::CoreId;
+use crate::symbol::Exchange;
+
+/// The catalog of one core's market-data provider, frozen at one snapshot.
+pub struct CatalogLookup {
+    snapshot: moonproto::MoonClientSnapshot,
+    exchange: Exchange,
+}
+
+impl MarketDataSource {
+    /// Freeze the catalog `core` reads its markets from, for lookups that must agree.
+    ///
+    /// Returns:
+    ///     The lookup, or `None` while the core has no provider, client or snapshot yet.
+    pub fn catalog_lookup(&self, core: CoreId) -> Option<CatalogLookup> {
+        let (client, exchange) = {
+            let inner = self.inner.read().expect("market source poisoned");
+            let provider = inner.core_provider.get(&core).copied()?;
+            let client = inner
+                .clients
+                .get(&provider)
+                .and_then(SharedMoonClient::get)?;
+            (client, inner.exchange_of_provider(provider))
+        };
+        Some(CatalogLookup {
+            snapshot: client.snapshot_versioned()?,
+            exchange,
+        })
+    }
+}
+
+impl CatalogLookup {
+    /// MoonProto's version of the market-name universe in this snapshot.
+    ///
+    /// It moves when new names become visible and may move on a full-list apply; it does not
+    /// move on prices, tags or trading status. Comparable only within one provider client.
+    pub fn markets_version(&self) -> u64 {
+        self.snapshot.markets().markets_version()
+    }
+
+    /// The catalog-verified market a report row's `coin` names, from this snapshot alone.
+    ///
+    /// The rules are [`MarketDataSource::resolve_market`]'s, which calls this.
+    ///
+    /// Args:
+    ///     quote: The core's quote setting (`ServerConfig::market`).
+    ///     traded_quote: The core's account currency, when it has stated one.
+    ///     coin: Historical base coin or full market value from the row.
+    ///
+    /// Returns:
+    ///     Catalog-verified market, or `None` when this catalog cannot resolve it.
+    pub fn resolve_market(
+        &self,
+        quote: &str,
+        traded_quote: Option<String>,
+        coin: &str,
+    ) -> Option<String> {
+        let exchange = self.exchange;
+        let quote = crate::symbol::resolve_quote_on(quote, exchange);
+        // A complete market is one carrying a HIP-3 DEX prefix or THIS CORE'S quote. Deliberately
+        // not "carries any recognized quote": a coin whose own name ends in one — `WBTC`,
+        // `STETH`, `PYUSD` — would then be taken for a finished market and never get its quote
+        // appended.
+        let parsed = crate::symbol::parse::split_market(coin, exchange);
+        let already_full = parsed.dex.is_some()
+            || (!quote.is_empty() && parsed.quote.eq_ignore_ascii_case(&quote));
+        let candidate = if already_full || quote.is_empty() {
+            coin.to_string()
+        } else {
+            crate::symbol::parse::market_names_for(coin, &quote, exchange)
+                .next()
+                .unwrap_or_else(|| coin.to_string())
+        };
+        let catalog = self.snapshot.markets();
+        let universe: Vec<String> = catalog
+            .search(coin, 32)
+            .into_iter()
+            .map(|handle| handle.name().to_string())
+            .collect();
+        if universe.is_empty() {
+            return None;
+        }
+        let labelled: Vec<(String, MarketLabel)> = universe
+            .iter()
+            .map(|market| (market.clone(), label_in(Some(catalog), market, exchange)))
+            .collect();
+        let traded_quote = traded_quote.unwrap_or_else(|| quote.clone());
+        if let Some(name) = crate::market::pick_market_for_coin(&labelled, coin, &traded_quote) {
+            return Some(name.to_string());
+        }
+        // The historical format where the stored value is already a full market name.
+        universe
+            .iter()
+            .find(|market| market.eq_ignore_ascii_case(&candidate))
+            .cloned()
+    }
+}
+
+/// How `market` is named to the user, from `catalog` first and from its name otherwise.
+///
+/// The one copy of [`MarketDataSource::market_labels`]'s per-market rule, shared with
+/// [`CatalogLookup::resolve_market`].
+pub(super) fn label_in(
+    catalog: Option<&MarketsState>,
+    market: &str,
+    exchange: Exchange,
+) -> MarketLabel {
+    catalog
+        .and_then(|catalog| catalog.get(market))
+        .and_then(|handle| {
+            handle.with(|m| {
+                let coin = m.market_currency.trim();
+                (!coin.is_empty()).then(|| {
+                    let quote = m.base_currency.trim().to_ascii_uppercase();
+                    // What the NAME alone says, read ONCE: both fields below fall back to it, and
+                    // this runs per market of a search result.
+                    let from_name = MarketLabel::from_name(market, exchange);
+                    MarketLabel {
+                        coin: coin.to_string(),
+                        // The core's own cross-exchange identity, carried whole. See
+                        // `MarketLabel::canonic` for why it is not derived here.
+                        canonic: m.market_currency_canonic.trim().to_string(),
+                        // A COIN-M contract reports no base currency at all, so its quote comes
+                        // from the name (`BTCUSD_PERP` → `USD`); without this the pair collapses
+                        // to a bare coin.
+                        quote: match quote.is_empty() {
+                            true => from_name.quote,
+                            false => quote,
+                        },
+                        // The expiry the NAME carries, which the token often does not: Bybit's
+                        // dated contract is the market `BTCUSDT-25SEP26` whose token is a flat
+                        // `BTCUSDT25SEP`, and reading only the token made every one of its ten
+                        // expiries look like a perpetual. COIN-M spells it inside the token
+                        // (`BTC_0925`) and answers from there, so this fills a gap rather than
+                        // competing.
+                        contract: from_name.contract,
+                    }
+                })
+            })
+        })
+        // A market the catalog does not hold — delisted under an open order, or a core that has
+        // not sent its list yet.
+        .unwrap_or_else(|| MarketLabel::from_name(market, exchange))
+}

crates/moon-core/src/market/source/mod.rs

@@ -2,11 +2,13 @@
 
 mod arb;
 mod archive;
+mod catalog_lookup;
 pub(crate) use archive::ARCHIVE_WAIT;
 /// How far back a core's chart archive serves trade prints.
 pub const CORE_ARCHIVE_RETENTION_MS: i64 = 48 * 3_600_000;
 mod history;
 mod open_path;
+pub use catalog_lookup::CatalogLookup;
 pub use open_path::PathPrefixRows;
 #[cfg(test)]
 mod label_tests;

crates/moon-core/src/market/source/read.rs

@@ -442,41 +442,9 @@ impl MarketDataSource {
     /// Returns:
     ///     Catalog-verified market, or `None` while the core cannot resolve it.
     pub fn resolve_market(&self, core: CoreId, quote: &str, coin: &str) -> Option<String> {
-        let exchange = self.exchange_of(core);
-        let quote = crate::symbol::resolve_quote_on(quote, exchange);
-        // A complete market is one carrying a HIP-3 DEX prefix or THIS CORE'S quote. Deliberately
-        // not "carries any recognized quote": a coin whose own name ends in one — `WBTC`,
-        // `STETH`, `PYUSD` — would then be taken for a finished market and never get its quote
-        // appended.
-        let parsed = crate::symbol::parse::split_market(coin, exchange);
-        let already_full = parsed.dex.is_some()
-            || (!quote.is_empty() && parsed.quote.eq_ignore_ascii_case(&quote));
-        let candidate = if already_full || quote.is_empty() {
-            coin.to_string()
-        } else {
-            crate::symbol::parse::market_names_for(coin, &quote, exchange)
-                .next()
-                .unwrap_or_else(|| coin.to_string())
-        };
-        let universe = self.search_markets(core, coin, 32);
-        if universe.is_empty() {
-            return None;
-        }
-        let refs: Vec<&str> = universe.iter().map(String::as_str).collect();
-        let labelled: Vec<(String, crate::market::MarketLabel)> = universe
-            .iter()
-            .cloned()
-            .zip(self.market_labels(core, &refs))
-            .collect();
-        let traded_quote = self.traded_quote(core).unwrap_or_else(|| quote.clone());
-        if let Some(name) = crate::market::pick_market_for_coin(&labelled, coin, &traded_quote) {
-            return Some(name.to_string());
-        }
-        // The historical format where the stored value is already a full market name.
-        universe
-            .iter()
-            .find(|market| market.eq_ignore_ascii_case(&candidate))
-            .cloned()
+        // One snapshot answers the search and the labels: `CatalogLookup` holds it.
+        self.catalog_lookup(core)?
+            .resolve_market(quote, self.traded_quote(core), coin)
     }
 
     /// The quote currency `core` trades in — its account currency from BaseCheck
@@ -541,45 +509,7 @@ impl MarketDataSource {
         let catalog = snapshot.as_ref().map(|snapshot| snapshot.markets());
         markets
             .iter()
-            .map(|market| {
-                catalog
-                    .as_ref()
-                    .and_then(|catalog| catalog.get(market))
-                    .and_then(|handle| {
-                        handle.with(|m| {
-                            let coin = m.market_currency.trim();
-                            (!coin.is_empty()).then(|| {
-                                let quote = m.base_currency.trim().to_ascii_uppercase();
-                                // What the NAME alone says, read ONCE: both fields below fall back
-                                // to it, and this runs per market of a search result.
-                                let from_name = MarketLabel::from_name(market, exchange);
-                                MarketLabel {
-                                    coin: coin.to_string(),
-                                    // The core's own cross-exchange identity, carried whole. See
-                                    // `MarketLabel::canonic` for why it is not derived here.
-                                    canonic: m.market_currency_canonic.trim().to_string(),
-                                    // A COIN-M contract reports no base currency at all, so its
-                                    // quote comes from the name (`BTCUSD_PERP` → `USD`); without
-                                    // this the pair collapses to a bare coin.
-                                    quote: match quote.is_empty() {
-                                        true => from_name.quote,
-                                        false => quote,
-                                    },
-                                    // The expiry the NAME carries, which the token often does not:
-                                    // Bybit's dated contract is the market `BTCUSDT-25SEP26` whose
-                                    // token is a flat `BTCUSDT25SEP`, and reading only the token
-                                    // made every one of its ten expiries look like a perpetual.
-                                    // COIN-M spells it inside the token (`BTC_0925`) and answers
-                                    // from there, so this fills a gap rather than competing.
-                                    contract: from_name.contract,
-                                }
-                            })
-                        })
-                    })
-                    // A market the catalog does not hold — delisted under an open order, or a
-                    // core that has not sent its list yet.
-                    .unwrap_or_else(|| MarketLabel::from_name(market, exchange))
-            })
+            .map(|market| super::catalog_lookup::label_in(catalog, market, exchange))
             .collect()
     }
 

crates/moon-core/src/session/lifecycle.rs

@@ -11,7 +11,7 @@ use crate::feed::{
     self, ConnStatus, CoreTgEvent, CoreTimeOffsetStatus, EngineActionResult, ExchangeId,
     FeedHandle, FeedMsg, FeedWakeTx,
 };
-use crate::market::{MarketDataMode, MarketDataSource, MarketStore};
+use crate::market::{CatalogLookup, MarketDataMode, MarketDataSource, MarketStore};
 use crate::session::core_time_offset::OffsetSource;
 use crate::venue::CoreVenue;
 
@@ -952,16 +952,27 @@ impl SessionManager {
                     let Some((open_ms, close_ms)) = catchup_stamps(row, &axis, core, now_ms) else {
                         continue;
                     };
+                    // One snapshot answers the row and versions the answer (`CoinMarkets`).
+                    // Reading it is paid on every row, hits too, so it counts as resolving.
+                    let reading = Instant::now();
+                    let lookup = source.catalog_lookup(core);
+                    let traded_quote = source.traded_quote(core);
+                    markets.spent += reading.elapsed();
                     let resolved = markets.resolve(
                         || {
                             (
+                                lookup.as_ref().map(CatalogLookup::markets_version),
                                 source.catalog_revision(core),
-                                source.traded_quote(core),
+                                traded_quote.clone(),
                                 source.exchange_of(core),
                             )
                         },
                         &row.coin,
-                        || source.resolve_market(core, &quote, &row.coin),
+                        || {
+                            lookup.as_ref().and_then(|lookup| {
+                                lookup.resolve_market(&quote, traded_quote.clone(), &row.coin)
+                            })
+                        },
                     );
                     let Some(market) = resolved else {
                         if source.tradable_coins(core).is_none() {