Use fallback oracles in Rust client (#838)

* rename usd_opt to usdc_opt in OracleAccountInfos

* use fallbacks when fetching bank+ price in AccountFetcher struct

* feat: add derive_fallback_oracle_keys to MangoGroupContext

* test: properly assert failure CU in test_health_compute_tokens_fallback_oracles

* provide fallback oracle accounts in the rust client

* liquidator: update for fallback oracles

* set fallback config in mango services

* support fallback oracles in rust settler + keeper

* fix send error related to fetching fallbacks dynamically in tcs_start

* drop derive_fallback_oracle_keys_sync

* add fetch_multiple_accounts to AccountFetcher trait

* revert client::new() api

* deriving oracle keys uses account_fetcher

* use client helpers for deriving health_check account_metas

* add health_cache helper to mango client

* add get_slot to account_fetcher

* lint

* cached account fetcher only fetches uncached accounts

* ensure keeper client does not use cached oracles for staleness checks

* address minor review comments

* create unique job keys for CachedAccountFetcher.fetch_multiple_accounts

* fmt

* improve hashing in CachedAccountFetcher.fetch_multiple_accounts

---------

Co-authored-by: Christian Kamm <mail@ckamm.de>

bin/keeper/src/main.rs

@@ -7,7 +7,9 @@ use std::time::Duration;
 use anchor_client::Cluster;
 
 use clap::{Parser, Subcommand};
-use mango_v4_client::{keypair_from_cli, Client, MangoClient, TransactionBuilderConfig};
+use mango_v4_client::{
+    keypair_from_cli, Client, FallbackOracleConfig, MangoClient, TransactionBuilderConfig,
+};
 use solana_sdk::commitment_config::CommitmentConfig;
 use solana_sdk::pubkey::Pubkey;
 use tokio::time;
@@ -98,19 +100,21 @@ async fn main() -> Result<(), anyhow::Error> {
 
     let mango_client = Arc::new(
         MangoClient::new_for_existing_account(
-            Client::new(
-                cluster,
-                commitment,
-                owner.clone(),
-                Some(Duration::from_secs(cli.timeout)),
-                TransactionBuilderConfig {
+            Client::builder()
+                .cluster(cluster)
+                .commitment(commitment)
+                .fee_payer(Some(owner.clone()))
+                .timeout(Duration::from_secs(cli.timeout))
+                .transaction_builder_config(TransactionBuilderConfig {
                     prioritization_micro_lamports: (cli.prioritization_micro_lamports > 0)
                         .then_some(cli.prioritization_micro_lamports),
                     compute_budget_per_instruction: None,
-                },
-            ),
+                })
+                .fallback_oracle_config(FallbackOracleConfig::Never)
+                .build()
+                .unwrap(),
             cli.mango_account,
-            owner.clone(),
+            owner,
         )
         .await?,
     );

bin/liquidator/src/liquidate.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 use itertools::Itertools;
 use mango_v4::health::{HealthCache, HealthType};
 use mango_v4::state::{MangoAccountValue, PerpMarketIndex, Side, TokenIndex, QUOTE_TOKEN_INDEX};
-use mango_v4_client::{chain_data, health_cache, MangoClient};
+use mango_v4_client::{chain_data, MangoClient};
 use solana_sdk::signature::Signature;
 
 use futures::{stream, StreamExt, TryStreamExt};
@@ -155,10 +155,7 @@ impl<'a> LiquidateHelper<'a> {
                 .await
                 .context("getting liquidator account")?;
             liqor.ensure_perp_position(*perp_market_index, QUOTE_TOKEN_INDEX)?;
-            let mut health_cache =
-                health_cache::new(&self.client.context, self.account_fetcher, &liqor)
-                    .await
-                    .context("health cache")?;
+            let mut health_cache = self.client.health_cache(&liqor).await.expect("always ok");
             let quote_bank = self
                 .client
                 .first_bank(QUOTE_TOKEN_INDEX)
@@ -589,7 +586,8 @@ pub async fn maybe_liquidate_account(
     let liqor_min_health_ratio = I80F48::from_num(config.min_health_ratio);
 
     let account = account_fetcher.fetch_mango_account(pubkey)?;
-    let health_cache = health_cache::new(&mango_client.context, account_fetcher, &account)
+    let health_cache = mango_client
+        .health_cache(&account)
         .await
         .context("creating health cache 1")?;
     let maint_health = health_cache.health(HealthType::Maint);
@@ -607,7 +605,8 @@ pub async fn maybe_liquidate_account(
     // This is -- unfortunately -- needed because the websocket streams seem to not
     // be great at providing timely updates to the account data.
     let account = account_fetcher.fetch_fresh_mango_account(pubkey).await?;
-    let health_cache = health_cache::new(&mango_client.context, account_fetcher, &account)
+    let health_cache = mango_client
+        .health_cache(&account)
         .await
         .context("creating health cache 2")?;
     if !health_cache.is_liquidatable() {

bin/liquidator/src/rebalance.rs

@@ -520,6 +520,7 @@ impl Rebalancer {
             };
             let counters = perp_pnl::fetch_top(
                 &self.mango_client.context,
+                &self.mango_client.client.config().fallback_oracle_config,
                 self.account_fetcher.as_ref(),
                 perp_position.market_index,
                 direction,

bin/liquidator/src/trigger_tcs.rs

@@ -11,7 +11,7 @@ use mango_v4::{
     i80f48::ClampToInt,
     state::{Bank, MangoAccountValue, TokenConditionalSwap, TokenIndex},
 };
-use mango_v4_client::{chain_data, health_cache, jupiter, MangoClient, TransactionBuilder};
+use mango_v4_client::{chain_data, jupiter, MangoClient, TransactionBuilder};
 
 use anyhow::Context as AnyhowContext;
 use solana_sdk::{signature::Signature, signer::Signer};
@@ -665,8 +665,9 @@ impl Context {
         liqee_old: &MangoAccountValue,
         tcs_id: u64,
     ) -> anyhow::Result<Option<PreparedExecution>> {
-        let fetcher = self.account_fetcher.as_ref();
-        let health_cache = health_cache::new(&self.mango_client.context, fetcher, liqee_old)
+        let health_cache = self
+            .mango_client
+            .health_cache(liqee_old)
             .await
             .context("creating health cache 1")?;
         if health_cache.is_liquidatable() {
@@ -685,7 +686,9 @@ impl Context {
             return Ok(None);
         }
 
-        let health_cache = health_cache::new(&self.mango_client.context, fetcher, &liqee)
+        let health_cache = self
+            .mango_client
+            .health_cache(&liqee)
             .await
             .context("creating health cache 2")?;
         if health_cache.is_liquidatable() {

bin/service-mango-pnl/src/main.rs

@@ -21,7 +21,8 @@ use fixed::types::I80F48;
 use mango_feeds_connector::metrics::*;
 use mango_v4::state::{MangoAccount, MangoAccountValue, PerpMarketIndex};
 use mango_v4_client::{
-    chain_data, health_cache, AccountFetcher, Client, MangoGroupContext, TransactionBuilderConfig,
+    chain_data, health_cache, AccountFetcher, Client, FallbackOracleConfig, MangoGroupContext,
+    TransactionBuilderConfig,
 };
 use solana_sdk::commitment_config::CommitmentConfig;
 use solana_sdk::{account::ReadableAccount, signature::Keypair};
@@ -52,7 +53,13 @@ async fn compute_pnl(
     account_fetcher: Arc<impl AccountFetcher>,
     account: &MangoAccountValue,
 ) -> anyhow::Result<Vec<(PerpMarketIndex, I80F48)>> {
-    let health_cache = health_cache::new(&context, account_fetcher.as_ref(), account).await?;
+    let health_cache = health_cache::new(
+        &context,
+        &FallbackOracleConfig::Dynamic,
+        account_fetcher.as_ref(),
+        account,
+    )
+    .await?;
 
     let pnls = account
         .active_perp_positions()

bin/settler/src/settle.rs

@@ -5,9 +5,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use mango_v4::accounts_zerocopy::KeyedAccountSharedData;
 use mango_v4::health::HealthType;
 use mango_v4::state::{OracleAccountInfos, PerpMarket, PerpMarketIndex};
-use mango_v4_client::{
-    chain_data, health_cache, MangoClient, PreparedInstructions, TransactionBuilder,
-};
+use mango_v4_client::{chain_data, MangoClient, PreparedInstructions, TransactionBuilder};
 use solana_sdk::address_lookup_table_account::AddressLookupTableAccount;
 use solana_sdk::commitment_config::CommitmentConfig;
 use solana_sdk::signature::Signature;
@@ -113,7 +111,8 @@ impl SettlementState {
                 continue;
             }
 
-            let health_cache = health_cache::new(&mango_client.context, account_fetcher, &account)
+            let health_cache = mango_client
+                .health_cache(&account)
                 .await
                 .context("creating health cache")?;
             let liq_end_health = health_cache.health(HealthType::LiquidationEnd);
@@ -304,11 +303,14 @@ impl<'a> SettleBatchProcessor<'a> {
     ) -> anyhow::Result<Option<Signature>> {
         let a_value = self.account_fetcher.fetch_mango_account(&account_a)?;
         let b_value = self.account_fetcher.fetch_mango_account(&account_b)?;
-        let new_ixs = self.mango_client.perp_settle_pnl_instruction(
-            self.perp_market_index,
-            (&account_a, &a_value),
-            (&account_b, &b_value),
-        )?;
+        let new_ixs = self
+            .mango_client
+            .perp_settle_pnl_instruction(
+                self.perp_market_index,
+                (&account_a, &a_value),
+                (&account_b, &b_value),
+            )
+            .await?;
         let previous = self.instructions.clone();
         self.instructions.append(new_ixs.clone());
 

bin/settler/src/tcs_start.rs

@@ -113,14 +113,18 @@ impl State {
             }
 
             // Clear newly created token positions, so the liqor account is mostly empty
-            for token_index in startable_chunk.iter().map(|(_, _, ti)| *ti).unique() {
+            let new_token_pos_indices = startable_chunk
+                .iter()
+                .map(|(_, _, ti)| *ti)
+                .unique()
+                .collect_vec();
+            for token_index in new_token_pos_indices {
                 let mint = mango_client.context.token(token_index).mint;
-                instructions.append(mango_client.token_withdraw_instructions(
-                    &liqor_account,
-                    mint,
-                    u64::MAX,
-                    false,
-                )?);
+                let ix = mango_client
+                    .token_withdraw_instructions(&liqor_account, mint, u64::MAX, false)
+                    .await?;
+
+                instructions.append(ix)
             }
 
             let txsig = match mango_client

lib/client/src/account_fetcher.rs

@@ -11,10 +11,14 @@ use anchor_lang::AccountDeserialize;
 
 use solana_client::nonblocking::rpc_client::RpcClient as RpcClientAsync;
 use solana_sdk::account::{AccountSharedData, ReadableAccount};
+use solana_sdk::hash::Hash;
+use solana_sdk::hash::Hasher;
 use solana_sdk::pubkey::Pubkey;
 
 use mango_v4::state::MangoAccountValue;
 
+use crate::gpa;
+
 #[async_trait::async_trait]
 pub trait AccountFetcher: Sync + Send {
     async fn fetch_raw_account(&self, address: &Pubkey) -> anyhow::Result<AccountSharedData>;
@@ -29,6 +33,13 @@ pub trait AccountFetcher: Sync + Send {
         program: &Pubkey,
         discriminator: [u8; 8],
     ) -> anyhow::Result<Vec<(Pubkey, AccountSharedData)>>;
+
+    async fn fetch_multiple_accounts(
+        &self,
+        keys: &[Pubkey],
+    ) -> anyhow::Result<Vec<(Pubkey, AccountSharedData)>>;
+
+    async fn get_slot(&self) -> anyhow::Result<u64>;
 }
 
 // Can't be in the trait, since then it would no longer be object-safe...
@@ -100,6 +111,17 @@ impl AccountFetcher for RpcAccountFetcher {
             .map(|(pk, acc)| (pk, acc.into()))
             .collect::<Vec<_>>())
     }
+
+    async fn fetch_multiple_accounts(
+        &self,
+        keys: &[Pubkey],
+    ) -> anyhow::Result<Vec<(Pubkey, AccountSharedData)>> {
+        gpa::fetch_multiple_accounts(&self.rpc, keys).await
+    }
+
+    async fn get_slot(&self) -> anyhow::Result<u64> {
+        Ok(self.rpc.get_slot().await?)
+    }
 }
 
 struct CoalescedAsyncJob<Key, Output> {
@@ -138,6 +160,8 @@ struct AccountCache {
     keys_for_program_and_discriminator: HashMap<(Pubkey, [u8; 8]), Vec<Pubkey>>,
 
     account_jobs: CoalescedAsyncJob<Pubkey, anyhow::Result<AccountSharedData>>,
+    multiple_accounts_jobs:
+        CoalescedAsyncJob<Hash, anyhow::Result<Vec<(Pubkey, AccountSharedData)>>>,
     program_accounts_jobs:
         CoalescedAsyncJob<(Pubkey, [u8; 8]), anyhow::Result<Vec<(Pubkey, AccountSharedData)>>>,
 }
@@ -261,4 +285,62 @@ impl<T: AccountFetcher + 'static> AccountFetcher for CachedAccountFetcher<T> {
             )),
         }
     }
+
+    async fn fetch_multiple_accounts(
+        &self,
+        keys: &[Pubkey],
+    ) -> anyhow::Result<Vec<(Pubkey, AccountSharedData)>> {
+        let fetch_job = {
+            let mut cache = self.cache.lock().unwrap();
+            let mut missing_keys: Vec<Pubkey> = keys
+                .iter()
+                .filter(|k| !cache.accounts.contains_key(k))
+                .cloned()
+                .collect();
+            if missing_keys.len() == 0 {
+                return Ok(keys
+                    .iter()
+                    .map(|pk| (*pk, cache.accounts.get(&pk).unwrap().clone()))
+                    .collect::<Vec<_>>());
+            }
+
+            let self_copy = self.clone();
+            missing_keys.sort();
+            let mut hasher = Hasher::default();
+            for key in missing_keys.iter() {
+                hasher.hash(key.as_ref());
+            }
+            let job_key = hasher.result();
+            cache
+                .multiple_accounts_jobs
+                .run_coalesced(job_key.clone(), async move {
+                    let result = self_copy
+                        .fetcher
+                        .fetch_multiple_accounts(&missing_keys)
+                        .await;
+                    let mut cache = self_copy.cache.lock().unwrap();
+                    cache.multiple_accounts_jobs.remove(&job_key);
+
+                    if let Ok(results) = result.as_ref() {
+                        for (key, account) in results {
+                            cache.accounts.insert(*key, account.clone());
+                        }
+                    }
+                    result
+                })
+        };
+
+        match fetch_job.get().await {
+            Ok(v) => Ok(v.clone()),
+            // Can't clone the stored error, so need to stringize it
+            Err(err) => Err(anyhow::format_err!(
+                "fetch error in CachedAccountFetcher: {:?}",
+                err
+            )),
+        }
+    }
+
+    async fn get_slot(&self) -> anyhow::Result<u64> {
+        self.fetcher.get_slot().await
+    }
 }