Struct InstrumentsCache 

pub struct InstrumentsCache {
    instruments: Arc<RwLock<HashMap<String, Instrument>>>,
    by_underlying: Arc<RwLock<HashMap<String, Vec<String>>>>,
    by_underlying_expiry: Arc<RwLock<HashMap<(String, u64), Vec<String>>>>,
    by_instrument_id: Arc<RwLock<HashMap<i32, String>>>,
    last_market_update_seq: Arc<RwLock<i64>>,
    last_snapshot_seq: Arc<RwLock<i64>>,
    sync_status: Arc<SyncStatus>,
    has_received_update: AtomicBool,
}

Cache for fast instrument lookups Provides O(1) access to instruments by symbol, underlying, expiry, and instrument id

Fields

instruments: Arc<RwLock<HashMap<String, Instrument>>>

Primary storage: symbol -> Instrument

by_underlying: Arc<RwLock<HashMap<String, Vec<String>>>>

Index: underlying -> Vec

by_underlying_expiry: Arc<RwLock<HashMap<(String, u64), Vec<String>>>>

Index: (underlying, expiry) -> Vec

by_instrument_id: Arc<RwLock<HashMap<i32, String>>>

Index: instrument id -> symbol

last_market_update_seq: Arc<RwLock<i64>>

Last processed market update sequence number

last_snapshot_seq: Arc<RwLock<i64>>

Sequence captured during last list_all() call for atomic snapshot consistency. This ensures snapshot_offsets() returns the same seq that was captured with the state.

sync_status: Arc<SyncStatus>

Sync status for readiness tracking

has_received_update: AtomicBool

Flag to track if we’ve received at least one update (for readiness)

Implementations

impl InstrumentsCache

pub fn new() -> Self

Create a new empty instruments cache

pub fn sync_status(&self) -> Arc<SyncStatus>

Get the sync status for readiness checks.

pub fn is_ready(&self) -> bool

Check if the cache is ready to serve requests.

pub fn snapshot_offsets( &self, ) -> Result<HashMap<String, HashMap<i32, i64>>, SnapshotError>

Get snapshot offsets for the market updates topic. Returns the last processed sequence number as an offset.

pub async fn apply_snapshot_offsets( &self, offsets: HashMap<String, HashMap<i32, i64>>, ) -> Result<(), SnapshotError>

Apply snapshot offsets after restore.

pub async fn update_seq(&self, seq: i64)

Update the last processed sequence number.

pub async fn initialize( self: Arc<Self>, diesel_db: &Arc<dyn BootstrapReader>, event_bus: Arc<dyn EventBusTrait>, ) -> Result<()>

Initialize the cache by loading instruments from DB and starting event listener (legacy, no shutdown support). For graceful shutdown support, use initialize_with_shutdown.

pub async fn initialize_with_shutdown( self: Arc<Self>, diesel_db: &Arc<dyn BootstrapReader>, _event_bus: Arc<dyn EventBusTrait>, shutdown_rx: Receiver<()>, ) -> Result<()>

async fn start_periodic_refresh( self: Arc<Self>, diesel_db: Arc<dyn BootstrapReader>, shutdown_rx: Receiver<()>, )

pub async fn initialize_with_snapshot<L>( self: Arc<Self>, diesel_db: &Arc<dyn BootstrapReader>, event_bus: Arc<dyn EventBusTrait>, loader: &L, shutdown_rx: Receiver<()>, ) -> Result<Option<HashMap<String, HashMap<i32, i64>>>>

where L: SnapshotLoader<Key = String, State = InstrumentSnapshotState>,

Initialize the cache with snapshot support.

Flow:

1. Try to load from snapshot (if available)
2. If snapshot exists: restore state, catchup from offsets, mark as ready
3. If no snapshot: load from DB as fallback, mark as ready
4. Start event listener for live updates

async fn load_instruments_from_db( &self, diesel_db: &dyn BootstrapReader, ) -> Result<()>

async fn start_event_listeners_with_shutdown( self: Arc<Self>, event_bus: Arc<dyn EventBusTrait>, shutdown_rx: Receiver<()>, ) -> Result<()>

Start event listeners for market updates with shutdown support. Uses subscribe_with_offsets to track offsets for snapshot consistency.

pub async fn handle_market_update( &self, message: MarketUpdateMessage, ) -> Result<()>

Handle market update messages (create, delete, expire) Public to allow catchup replay from saved offsets.

async fn add_instrument(&self, market: &Market) -> Result<()>

Add a new instrument to the cache

async fn remove_instrument(&self, symbol: &str) -> Result<()>

Remove an instrument from the cache

pub async fn get_by_symbol(&self, symbol: &str) -> Option<Instrument>

Get instrument by symbol

pub fn allows_rfq(&self, symbol: &str) -> bool

Synchronously check whether an instrument’s trading mode includes RFQ. Returns false for unknown instruments so callers default to rejecting.

pub async fn get_by_underlying(&self, underlying: &str) -> Vec<Instrument>

Get all instruments for a given underlying

pub async fn get_by_underlying_and_expiry( &self, underlying: &str, expiry: u64, ) -> Vec<Instrument>

Get all instruments for a given underlying and expiry

pub async fn get_by_instrument_id( &self, instrument_id: i32, ) -> Option<Instrument>

Get instrument by instrument id

pub async fn get_all(&self) -> Vec<Instrument>

Get all instruments (for debugging/admin)

pub async fn len(&self) -> usize

Get count of cached instruments

pub async fn is_empty(&self) -> bool

Check if cache is empty

pub async fn reload_from_db( &self, diesel_db: &dyn BootstrapReader, ) -> Result<()>

async fn add_instrument_direct(&self, instrument: Instrument) -> Result<()>

Add an instrument directly to the cache (used during snapshot restore).

Trait Implementations

impl Default for InstrumentsCache

fn default() -> Self

Returns the “default value” for a type.

impl InstrumentsCacheReader for InstrumentsCache

fn get_by_symbol<'life0, 'life1, 'async_trait>( &'life0 self, symbol: &'life1 str, ) -> Pin<Box<dyn Future<Output = Option<Instrument>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn allows_rfq(&self, symbol: &str) -> bool

fn get_by_underlying_and_expiry<'life0, 'life1, 'async_trait>( &'life0 self, underlying: &'life1 str, expiry: u64, ) -> Pin<Box<dyn Future<Output = Vec<Instrument>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn get_by_instrument_id<'life0, 'async_trait>( &'life0 self, instrument_id: i32, ) -> Pin<Box<dyn Future<Output = Option<Instrument>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn get_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Vec<Instrument>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn len<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = usize> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn reload_from_db<'life0, 'life1, 'async_trait>( &'life0 self, db: &'life1 dyn BootstrapReader, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

impl Snapshotable for InstrumentsCache

fn list_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<HashMap<Self::Key, Self::State>, SnapshotError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns a cloned snapshot of all instrument entries.

CRITICAL: We must capture the state AND the sequence atomically. This ensures snapshot_offsets() returns the same seq that was captured with the state.

type Key = String

The key type for identifying entities (e.g., WalletAddress)

type State = InstrumentSnapshotState

The state type for each keyed entity (e.g., wallet -> balance)

fn restore<'life0, 'life1, 'async_trait>( &'life0 self, key: &'life1 Self::Key, state: Self::State, ) -> Pin<Box<dyn Future<Output = Result<(), SnapshotError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Restore a single state entry.

fn clear_all<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), SnapshotError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Clear all state before restore.