Merge pull request #102 from 0dotxyz/1.9

1.9

Author: IliaZyrin

src/cache/banks.rs

@@ -3,11 +3,9 @@ use crate::{
     wrappers::bank::BankWrapper,
 };
 use anyhow::{anyhow, Result};
+use marginfi::utils::is_marginfi_asset_tag;
 use marginfi_type_crate::{
-    constants::{
-        ASSET_TAG_DEFAULT, ASSET_TAG_DRIFT, ASSET_TAG_JUPLEND, ASSET_TAG_KAMINO, ASSET_TAG_SOL,
-        ASSET_TAG_STAKED,
-    },
+    constants::{ASSET_TAG_DRIFT, ASSET_TAG_JUPLEND, ASSET_TAG_KAMINO},
     types::{Bank, OracleSetup},
 };
 use solana_sdk::{account::Account, pubkey::Pubkey};
@@ -54,10 +52,7 @@ impl BanksCache {
         inner
             .banks
             .insert(bank_address, BankWrapper::new(bank_address, bank, account));
-        if matches!(
-            bank.config.asset_tag,
-            ASSET_TAG_DEFAULT | ASSET_TAG_SOL | ASSET_TAG_STAKED
-        ) {
+        if is_marginfi_asset_tag(bank.config.asset_tag) {
             inner.mint_to_p0_bank.insert(bank.mint, bank_address);
         }
         Ok(())

src/cache_loader.rs

@@ -295,7 +295,10 @@ impl CacheLoader {
         // data over this placeholder.
         for onramp in cache.banks.get_staked_onramps() {
             if !oracle_map.contains_key(&onramp) {
-                debug!("Inserting empty placeholder for staked on-ramp {:?}.", onramp);
+                debug!(
+                    "Inserting empty placeholder for staked on-ramp {:?}.",
+                    onramp
+                );
                 cache.oracles.try_insert(onramp, Account::default())?;
             }
         }

src/geyser.rs

@@ -20,6 +20,8 @@ use yellowstone_grpc_client::{ClientTlsConfig, GeyserGrpcClient};
 use yellowstone_grpc_proto::prelude::*;
 
 const RATE_LIMIT_LOG_INTERVAL_SECS: u64 = 60;
+const INITIAL_RECONNECT_BACKOFF: Duration = Duration::from_secs(1);
+const MAX_RECONNECT_BACKOFF: Duration = Duration::from_secs(30);
 
 #[derive(Debug, Clone)]
 pub struct GeyserUpdate {
@@ -112,6 +114,7 @@ impl GeyserService {
         let tracked_accounts_vec: Vec<Pubkey> = self.tracked_accounts.keys().copied().collect();
         let tls_config = ClientTlsConfig::new().with_native_roots();
         let mut from_slot: Option<u64> = None;
+        let mut backoff = INITIAL_RECONNECT_BACKOFF;
 
         while !self.stop.load(Ordering::Relaxed) {
             info!("Connecting to Geyser...");
@@ -124,21 +127,43 @@ impl GeyserService {
 
             // TODO: replace from_slot with auto-reconnect once we migrate to the up-to-date client (requires updating Solana deps):
             // https://docs.triton.one/project-yellowstone/dragons-mouth-grpc-subscriptions#auto-reconnect-rust-client
-            let mut client = self.tokio_rt.block_on(
-                GeyserGrpcClient::build_from_shared(self.endpoint.clone())?
+            //
+            // Establish the connection and subscription inside the reconnect loop so that a
+            // transient Geyser outage (e.g. connection refused) triggers a retry with backoff
+            // instead of propagating out of `start()` and panicking the whole process.
+            let connect_result = self.tokio_rt.block_on(async {
+                let mut client = GeyserGrpcClient::build_from_shared(self.endpoint.clone())?
                     .x_token(self.x_token.clone())?
                     .tls_config(tls_config.clone())?
-                    .connect(),
-            )?;
+                    .connect()
+                    .await?;
+                let (_, stream) = client.subscribe_with_request(Some(sub_req.clone())).await?;
+                Ok::<_, anyhow::Error>(stream)
+            });
+
+            let mut stream = match connect_result {
+                // Don't reset the backoff yet: a server can accept the connection and then
+                // immediately reset the stream (REFUSED_STREAM). Only treat the connection as
+                // healthy once it actually delivers a message (see below).
+                Ok(stream) => stream,
+                Err(e) => {
+                    self.error_logger.warn(&format!(
+                        "Failed to connect to Geyser, retrying in {:?}: {:?}",
+                        backoff, e
+                    ));
+                    self.sleep_interruptible(backoff);
+                    backoff = (backoff * 2).min(MAX_RECONNECT_BACKOFF);
+                    continue;
+                }
+            };
 
-            let (_, mut stream) = self
-                .tokio_rt
-                .block_on(client.subscribe_with_request(Some(sub_req.clone())))?;
             // TODO: use IndexerFlags
             info!("Entering the GeyserService loop");
             while let Some(msg) = self.tokio_rt.block_on(stream.next()) {
                 match msg {
                     Ok(msg) => {
+                        // A delivered message proves the connection is healthy: reset the backoff.
+                        backoff = INITIAL_RECONNECT_BACKOFF;
                         let update_oneof = ward!(msg.update_oneof, continue);
                         if let subscribe_update::UpdateOneof::Account(account) = update_oneof {
                             from_slot = Some(account.slot);
@@ -176,10 +201,15 @@ impl GeyserService {
                     }
                     Err(error) => {
                         self.error_logger.warn(&format!(
-                            "Received error message from Geyser, reconnecting: {:?}",
-                            error
+                            "Received error message from Geyser, reconnecting in {:?}: {:?}",
+                            backoff, error
                         ));
 
+                        // Back off before reconnecting so a server that keeps resetting the
+                        // stream isn't hammered in a tight loop.
+                        self.sleep_interruptible(backoff);
+                        backoff = (backoff * 2).min(MAX_RECONNECT_BACKOFF);
+
                         // Break the inner loop so the outer loop reconnects.
                         break;
                     }
@@ -196,6 +226,15 @@ impl GeyserService {
         Ok(())
     }
 
+    /// Sleeps up to `duration`, waking early if a stop is requested so the reconnect
+    /// backoff never delays a clean shutdown.
+    fn sleep_interruptible(&self, duration: Duration) {
+        let deadline = Instant::now() + duration;
+        while Instant::now() < deadline && !self.stop.load(Ordering::Relaxed) {
+            std::thread::sleep(Duration::from_millis(200).min(deadline - Instant::now()));
+        }
+    }
+
     fn send_update(&self, account_type: AccountType, address: Pubkey, account: &Account) {
         let update = GeyserUpdate {
             account_type,

src/liquidator.rs

@@ -50,7 +50,6 @@ use std::{
 const MAX_CONCURRENT_LIQUIDATIONS: usize = 8;
 
 pub struct Liquidator {
-    liquidator_account: Arc<LiquidatorAccount>,
     rebalancer: Rebalancer,
     executor: Executor,
     strategy: InventoryStrategy,
@@ -121,7 +120,6 @@ impl Liquidator {
         )?;
 
         Ok(Liquidator {
-            liquidator_account,
             rebalancer,
             executor,
             strategy,
@@ -314,10 +312,6 @@ impl Liquidator {
         let mut result: Vec<PreparedLiquidatableAccount> = vec![];
 
         for account_address in account_addresses {
-            if account_address == self.liquidator_account.liquidator_address {
-                continue;
-            }
-
             let account = self
                 .cache
                 .marginfi_accounts

src/marginfi_ixs.rs

@@ -1,7 +1,7 @@
 use anchor_lang::{Id, InstructionData, Key, ToAccountMetas};
 
 use anchor_spl::{associated_token, token_2022};
-use log::{debug, info, trace};
+use log::{debug, trace};
 use marginfi_type_crate::{
     constants::LIQUIDATION_RECORD_SEED,
     pdas::{
@@ -10,13 +10,9 @@ use marginfi_type_crate::{
         derive_kamino_user_state,
     },
 };
-use solana_client::{rpc_client::RpcClient, rpc_config::RpcSendTransactionConfig};
-use solana_commitment_config::CommitmentConfig;
 use solana_sdk::{
     instruction::{AccountMeta, Instruction},
     pubkey::Pubkey,
-    signature::Keypair,
-    signer::Signer,
     sysvar,
 };
 use solana_sdk_ids::system_program;
@@ -213,25 +209,6 @@ fn maybe_add_bank_mint(accounts: &mut Vec<AccountMeta>, mint: Pubkey, token_prog
     }
 }
 
-pub fn make_create_ix(
-    marginfi_group: Pubkey,
-    marginfi_account: Pubkey,
-    signer: Pubkey,
-) -> Instruction {
-    Instruction {
-        program_id: marginfi_type_crate::ID,
-        accounts: marginfi::accounts::MarginfiAccountInitialize {
-            marginfi_group,
-            marginfi_account,
-            system_program: solana_sdk_ids::system_program::ID,
-            authority: signer,
-            fee_payer: signer,
-        }
-        .to_account_metas(Some(true)),
-        data: marginfi::instruction::MarginfiAccountInitialize.data(),
-    }
-}
-
 fn mark_signer(
     accounts: &mut [solana_sdk::instruction::AccountMeta],
     signer: solana_sdk::pubkey::Pubkey,
@@ -241,44 +218,6 @@ fn mark_signer(
     }
 }
 
-pub fn initialize_marginfi_account(
-    rpc_client: &RpcClient,
-    marginfi_group: Pubkey,
-    signer_keypair: &Keypair,
-) -> anyhow::Result<Pubkey> {
-    let marginfi_account_key = Keypair::new();
-
-    let ix = make_create_ix(
-        marginfi_group,
-        marginfi_account_key.pubkey(),
-        signer_keypair.pubkey(),
-    );
-
-    let recent_blockhash = rpc_client.get_latest_blockhash()?;
-    let tx = solana_sdk::transaction::Transaction::new_signed_with_payer(
-        &[ix],
-        Some(&signer_keypair.pubkey()),
-        &[signer_keypair, &marginfi_account_key],
-        recent_blockhash,
-    );
-
-    let res = rpc_client.send_and_confirm_transaction_with_spinner_and_config(
-        &tx,
-        CommitmentConfig::finalized(),
-        RpcSendTransactionConfig {
-            skip_preflight: true,
-            ..Default::default()
-        },
-    );
-    info!(
-        "Initialized new Marginfi account {:?} (without preflight check): {:?} ",
-        marginfi_account_key.pubkey(),
-        res
-    );
-
-    Ok(marginfi_account_key.pubkey())
-}
-
 #[allow(clippy::too_many_arguments)]
 pub fn make_kamino_withdraw_ix(
     group: Pubkey,

src/rebalancer.rs

@@ -12,7 +12,7 @@ use std::{collections::HashSet, sync::Arc};
 use tokio::runtime::{Builder, Runtime};
 
 /// Don't bother selling a position worth less than this (USD); the swap fee/dust isn't worth it.
-const MIN_REBALANCE_VALUE: I80F48 = I80F48!(0.5);
+const MIN_REBALANCE_VALUE: I80F48 = I80F48!(1.0);
 
 /// The rebalancer keeps the liquidator holding only the swap mint (USDC): every other token it ends
 /// up with — seized collateral from a liquidation, or a JIT-buy overshoot — is sold back to USDC on

src/wrappers/liquidator_account.rs

@@ -6,9 +6,9 @@ use crate::{
     juplend_ixs::make_update_lending_rate_ix,
     kamino_ixs::{make_refresh_obligation_ix, make_refresh_reserve_ix},
     marginfi_ixs::{
-        initialize_marginfi_account, make_drift_withdraw_ix, make_end_liquidate_ix,
-        make_init_liquidation_record_ix, make_juplend_withdraw_ix, make_kamino_withdraw_ix,
-        make_repay_ix, make_start_liquidate_ix, make_withdraw_ix,
+        make_drift_withdraw_ix, make_end_liquidate_ix, make_init_liquidation_record_ix,
+        make_juplend_withdraw_ix, make_kamino_withdraw_ix, make_repay_ix, make_start_liquidate_ix,
+        make_withdraw_ix,
     },
     utils::{self, marginfi_account_by_authority},
 };
@@ -35,7 +35,7 @@ use solana_sdk::{
     signer::Signer,
     transaction::VersionedTransaction,
 };
-use std::{collections::HashSet, sync::Arc, thread, time::Duration};
+use std::{collections::HashSet, sync::Arc};
 
 pub const PROFIT_SHARE: f64 = 0.085;
 
@@ -53,7 +53,6 @@ pub struct PreparedLiquidatableAccount {
 }
 
 pub struct LiquidatorAccount {
-    pub liquidator_address: Pubkey,
     pub signer: Keypair,
     group: Pubkey,
     rpc_client: RpcClient,
@@ -79,27 +78,7 @@ impl LiquidatorAccount {
             accounts
         );
 
-        let liquidator_address = if accounts.is_empty() {
-            info!("No MarginFi account found for the provided signer. Creating it...");
-            let liquidator_marginfi_account =
-                initialize_marginfi_account(&rpc_client, marginfi_group_id, &signer)?;
-
-            while cache
-                .marginfi_accounts
-                .try_get_account(&liquidator_marginfi_account)
-                .is_err()
-            {
-                info!("Waiting for the new account info to arrive...");
-                thread::sleep(Duration::from_secs(5));
-            }
-
-            liquidator_marginfi_account
-        } else {
-            accounts[0]
-        };
-
         Ok(Self {
-            liquidator_address,
             signer,
             group: marginfi_group_id,
             rpc_client,
@@ -113,8 +92,8 @@ impl LiquidatorAccount {
 
     pub fn init_liq_record(&self, liquidatee_account: &MarginfiAccountWrapper) -> Result<Pubkey> {
         info!(
-            "Initializing liquidation record for account {:?} with liquidator account {:?}.",
-            liquidatee_account.address, self.liquidator_address
+            "Initializing liquidation record for account {:?}",
+            liquidatee_account.address
         );
 
         let signer_pk = self.signer.pubkey();