cap the number of concurrent Jupiter calls to 2 to avoid 429 errors

Author: IliaZyrin

Cargo.toml

@@ -54,7 +54,7 @@ serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.149"
 signal-hook = "0.4.3"
 tiny_http = "0.12.0"
-tokio = { version = "1.50.0", features = ["rt", "macros"] }
+tokio = { version = "1.50.0", features = ["rt", "macros", "sync"] }
 warp = "0.4.2"
 
 yellowstone-grpc-client = "10.2.0"

src/execution/inventory.rs

@@ -22,7 +22,10 @@ use log::{debug, info};
 use solana_dex_superagg::{buy_shortfall, DexSuperAggClient};
 use solana_program::pubkey::Pubkey;
 use solana_sdk::transaction::VersionedTransaction;
-use tokio::runtime::{Builder, Runtime};
+use tokio::{
+    runtime::{Builder, Runtime},
+    sync::Semaphore,
+};
 
 use crate::clock_manager;
 use crate::wrappers::{
@@ -41,6 +44,7 @@ const INPUT_BUFFER: f64 = 1.02;
 pub struct InventoryStrategy {
     liquidator_account: Arc<LiquidatorAccount>,
     dex_client: Arc<DexSuperAggClient>,
+    dex_gate: Arc<Semaphore>,
     tokio_rt: Runtime,
     swap_mint: Pubkey,
     slippage_bps: u16,
@@ -51,6 +55,7 @@ impl InventoryStrategy {
         liquidator_account: Arc<LiquidatorAccount>,
         swap_mint: Pubkey,
         dex_client: Arc<DexSuperAggClient>,
+        dex_gate: Arc<Semaphore>,
         slippage_bps: u16,
     ) -> Result<Self> {
         // Multi-threaded so concurrent liquidations can `block_on` DEX quotes at the same time —
@@ -64,6 +69,7 @@ impl InventoryStrategy {
 
         Ok(Self {
             liquidator_account,
+            dex_gate,
             dex_client,
             tokio_rt,
             swap_mint,
@@ -114,16 +120,20 @@ impl InventoryStrategy {
         route_config.slippage_bps = Some(self.slippage_bps);
         route_config.wrap_and_unwrap_sol = false;
 
-        let prepared = self.tokio_rt.block_on(
+        // One permit for the whole sequence: the min-out flow quotes, re-quotes and then builds,
+        // which is a handful of aggregator requests per liquidation.
+        let prepared = self.tokio_rt.block_on(async {
+            let _permit = self.dex_gate.acquire().await?;
             self.dex_client
                 .build_swap_transaction_for_min_out_with_route_config(
                     &self.swap_mint.to_string(),
                     &output_mint.to_string(),
                     input_amount,
                     min_out,
                     route_config,
-                ),
-        )?;
+                )
+                .await
+        })?;
         debug!(
             "InventoryStrategy: ExactIn spends {} {} for {} {} (need {})",
             prepared.in_amount, self.swap_mint, prepared.out_amount, output_mint, min_out

src/liquidator.rs

@@ -93,6 +93,7 @@ impl Liquidator {
 
         let rebalancer = Rebalancer::new(config.clone(), cache.clone())?;
         let dex_client = rebalancer.dex_client();
+        let dex_gate = rebalancer.dex_gate();
 
         let jito = JitoClient::new(
             config.jito_block_engine_url.clone(),
@@ -116,6 +117,7 @@ impl Liquidator {
             liquidator_account.clone(),
             config.swap_mint,
             dex_client,
+            dex_gate,
             config.slippage_bps,
         )?;
 

src/rebalancer.rs

@@ -9,7 +9,15 @@ use solana_dex_superagg::{
 };
 use solana_program::pubkey::Pubkey;
 use std::{collections::HashSet, sync::Arc};
-use tokio::runtime::{Builder, Runtime};
+use tokio::{
+    runtime::{Builder, Runtime},
+    sync::Semaphore,
+};
+
+/// Concurrent DEX operations allowed across the liquidation workers and the rebalancer. Each one
+/// costs several aggregator round-trips, so an unbounded burst trips the Jupiter rate limit (429)
+/// and the failing targets then sit in the executor's backoff.
+const MAX_CONCURRENT_DEX_CALLS: usize = 2;
 
 /// 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!(1.0);
@@ -23,6 +31,7 @@ pub struct Rebalancer {
     tokio_rt: Runtime,
     cache: Arc<Cache>,
     dex_client: Arc<DexSuperAggClient>,
+    dex_gate: Arc<Semaphore>,
     empty_stake_banks: HashSet<Pubkey>,
     unpriceable_mints: HashSet<Pubkey>,
 }
@@ -74,6 +83,7 @@ impl Rebalancer {
             tokio_rt,
             cache,
             dex_client,
+            dex_gate: Arc::new(Semaphore::new(MAX_CONCURRENT_DEX_CALLS)),
             empty_stake_banks: HashSet::new(),
             unpriceable_mints: HashSet::new(),
         })
@@ -83,6 +93,10 @@ impl Rebalancer {
         Arc::clone(&self.dex_client)
     }
 
+    pub fn dex_gate(&self) -> Arc<Semaphore> {
+        Arc::clone(&self.dex_gate)
+    }
+
     /// Sell every non-swap-mint token the wallet holds (above the dust floor) back to the swap mint.
     pub fn run(&mut self) -> anyhow::Result<()> {
         info!("Running the Rebalancing process...");
@@ -191,12 +205,17 @@ impl Rebalancer {
         const WSOL: Pubkey = Pubkey::from_str_const("So11111111111111111111111111111111111111112");
         let wrap_and_unwrap_sol = input_mint == WSOL || output_mint == WSOL;
 
-        let result = self.tokio_rt.block_on(self.dex_client.swap(
-            &input_mint.to_string(),
-            &output_mint.to_string(),
-            amount,
-            wrap_and_unwrap_sol,
-        ))?;
+        let result = self.tokio_rt.block_on(async {
+            let _permit = self.dex_gate.acquire().await?;
+            self.dex_client
+                .swap(
+                    &input_mint.to_string(),
+                    &output_mint.to_string(),
+                    amount,
+                    wrap_and_unwrap_sol,
+                )
+                .await
+        })?;
 
         info!(
             "Swap successful! Transaction: {}, Output: {} tokens of mint {}",