fix: reduce forester load and update proving key downloads

cli/src/utils/processProverServer.ts

@@ -17,6 +17,7 @@ import {
 const LIGHT_CONFIG_DIR = path.join(os.homedir(), ".config", "light");
 const PROVER_BIN_DIR = path.join(LIGHT_CONFIG_DIR, "bin");
 const KEYS_DIR = path.join(LIGHT_CONFIG_DIR, "proving-keys");
+const PROVING_KEYS_BASE_URL = "https://d1wbn9ra8wjh7t.cloudfront.net";
 
 export async function killProver() {
   await killProcess(getProverNameByArch());
@@ -94,7 +95,8 @@ export async function startProver(proverPort: number, redisUrl?: string) {
 
   args.push("--keys-dir", KEYS_DIR + "/");
   args.push("--prover-address", `0.0.0.0:${proverPort}`);
-  args.push("--auto-download", "true");
+  args.push("--auto-download");
+  args.push("--download-url", PROVING_KEYS_BASE_URL);
 
   if (redisUrl) {
     args.push("--redis-url", redisUrl);

forester-utils/src/utils.rs

@@ -5,8 +5,7 @@ use light_client::{
     rpc::{Rpc, RpcError},
 };
 use solana_sdk::{signature::Signer, transaction::Transaction};
-use tokio::time::sleep;
-use tracing::{error, warn};
+use tracing::error;
 
 use crate::error::ForesterUtilsError;
 
@@ -30,10 +29,7 @@ pub async fn airdrop_lamports<R: Rpc>(
 
 pub async fn wait_for_indexer<R: Rpc>(rpc: &R) -> Result<(), ForesterUtilsError> {
     let rpc_slot = rpc.get_slot().await?;
-
-    let indexer_slot = rpc.indexer()?.get_indexer_slot(None).await;
-
-    let mut indexer_slot = match indexer_slot {
+    let indexer_slot = match rpc.indexer()?.get_indexer_slot(None).await {
         Ok(slot) => slot,
         Err(e) => {
             error!("failed to get indexer slot from indexer: {:?}", e);
@@ -43,32 +39,15 @@ pub async fn wait_for_indexer<R: Rpc>(rpc: &R) -> Result<(), ForesterUtilsError>
         }
     };
 
-    let max_attempts = 100;
-    let mut attempts = 0;
-
-    while rpc_slot > indexer_slot {
-        if attempts >= max_attempts {
-            return Err(ForesterUtilsError::Indexer(
-                "Maximum attempts reached waiting for indexer to catch up".into(),
-            ));
-        }
-
-        if rpc_slot - indexer_slot > 50 {
-            warn!(
-                "indexer is behind {} slots (rpc_slot: {}, indexer_slot: {})",
-                rpc_slot - indexer_slot,
-                rpc_slot,
-                indexer_slot
-            );
-        }
-
-        sleep(std::time::Duration::from_millis(1000)).await;
-        indexer_slot = rpc.indexer()?.get_indexer_slot(None).await.map_err(|e| {
-            error!("failed to get indexer slot from indexer: {:?}", e);
-            ForesterUtilsError::Indexer("Failed to get indexer slot".into())
-        })?;
-
-        attempts += 1;
+    let max_lag_slots = std::env::var("INDEXER_MAX_LAG_SLOTS")
+        .ok()
+        .and_then(|value| value.parse::<u64>().ok())
+        .unwrap_or(100);
+    let lag = rpc_slot.saturating_sub(indexer_slot);
+    if lag > max_lag_slots {
+        return Err(ForesterUtilsError::Indexer(format!(
+            "Indexer is behind {lag} slots (maximum allowed: {max_lag_slots})"
+        )));
     }
     Ok(())
 }

forester/src/api_server.rs

@@ -1118,15 +1118,23 @@ pub fn spawn_api_server(config: ApiServerConfig) -> anyhow::Result<ApiServerHand
                 run_id.clone(),
             ));
 
-            tokio::spawn(run_compressible_provider(
-                compressible_tx,
-                trackers,
-                config.forester_api_urls.clone(),
-                config.rpc_url.clone(),
-                provider_shutdown_tx.subscribe(),
-                config.helius_rpc,
-                run_id.clone(),
-            ));
+            // Only run the compressible-count provider when there is a CHEAP source:
+            // in-memory trackers (compression enabled) or upstream forester APIs to
+            // aggregate. With neither, it falls back to a full getProgramAccounts scan
+            // every 30s, which ties up RPC pool connections and triggers pool failures
+            // on mainnet. When compressible tracking is disabled (the default) skip it
+            // entirely — the dashboard simply reports no compressible counts.
+            if trackers.is_some() || !config.forester_api_urls.is_empty() {
+                tokio::spawn(run_compressible_provider(
+                    compressible_tx,
+                    trackers,
+                    config.forester_api_urls.clone(),
+                    config.rpc_url.clone(),
+                    provider_shutdown_tx.subscribe(),
+                    config.helius_rpc,
+                    run_id.clone(),
+                ));
+            }
 
             let cors = warp::cors()
                 .allow_any_origin()

forester/src/cli.rs

@@ -79,6 +79,14 @@ pub struct StartArgs {
     )]
     pub prover_max_wait_time_secs: Option<u64>,
 
+    #[arg(
+        long,
+        env = "PROVER_MAX_CONCURRENT_JOBS",
+        default_value = "4",
+        help = "Maximum queued or running prover jobs per forester (default: 4)"
+    )]
+    pub prover_max_concurrent_jobs: usize,
+
     #[arg(long, env = "PAYER")]
     pub payer: Option<String>,
 
@@ -168,7 +176,7 @@ pub struct StartArgs {
     )]
     pub priority_fee_microlamports: Option<u64>,
 
-    #[arg(long, env = "RPC_POOL_SIZE", default_value = "100")]
+    #[arg(long, env = "RPC_POOL_SIZE", default_value = "32")]
     pub rpc_pool_size: u32,
 
     #[arg(long, env = "RPC_POOL_CONNECTION_TIMEOUT_SECS", default_value = "15")]
@@ -303,6 +311,14 @@ pub struct StartArgs {
     )]
     pub enable_v1_multi_nullify: bool,
 
+    #[arg(
+        long,
+        env = "ENABLE_V1_PRESORT",
+        help = "Fetch queue leaf indices from the indexer (get_queue_leaf_indices) to pre-sort V1 work items for better dedup grouping. Requires an indexer that implements the endpoint. Best-effort; disabled by default.",
+        default_value = "false"
+    )]
+    pub enable_v1_presort: bool,
+
     #[arg(
         long,
         env = "WORK_ITEM_BATCH_SIZE",

forester/src/config.rs

@@ -36,6 +36,10 @@ pub struct ForesterConfig {
     /// Enable nullify_state_v1_multi instruction for batching 2-4 V1 state nullifications.
     /// Requires lookup_table_address to be set.
     pub enable_v1_multi_nullify: bool,
+    /// Enable the get_queue_leaf_indices pre-sort path for V1 work items.
+    /// Best-effort optimization; disabled by default. Only enable against an indexer
+    /// that implements the endpoint.
+    pub enable_v1_presort: bool,
     /// Number of queue items to process per batch cycle. Default: 50.
     pub work_item_batch_size: usize,
 }
@@ -52,6 +56,7 @@ pub struct ExternalServicesConfig {
     pub prover_api_key: Option<String>,
     pub prover_polling_interval: Option<Duration>,
     pub prover_max_wait_time: Option<Duration>,
+    pub prover_max_concurrent_jobs: usize,
     pub photon_grpc_url: Option<String>,
     pub pushgateway_url: Option<String>,
     pub pagerduty_routing_key: Option<String>,
@@ -284,6 +289,7 @@ impl ForesterConfig {
                 prover_api_key: args.prover_api_key.clone(),
                 prover_polling_interval: args.prover_polling_interval_ms.map(Duration::from_millis),
                 prover_max_wait_time: args.prover_max_wait_time_secs.map(Duration::from_secs),
+                prover_max_concurrent_jobs: args.prover_max_concurrent_jobs,
                 photon_grpc_url: args.photon_grpc_url.clone(),
                 pushgateway_url: args.push_gateway_url.clone(),
                 pagerduty_routing_key: args.pagerduty_routing_key.clone(),
@@ -431,6 +437,7 @@ impl ForesterConfig {
                 .transpose()?,
             min_queue_items: args.min_queue_items,
             enable_v1_multi_nullify: args.enable_v1_multi_nullify,
+            enable_v1_presort: args.enable_v1_presort,
             work_item_batch_size: args.work_item_batch_size.unwrap_or(50) as usize,
         })
     }
@@ -450,6 +457,7 @@ impl ForesterConfig {
                 prover_api_key: None,
                 prover_polling_interval: None,
                 prover_max_wait_time: None,
+                prover_max_concurrent_jobs: 4,
                 photon_grpc_url: None,
                 pushgateway_url: args.push_gateway_url.clone(),
                 pagerduty_routing_key: args.pagerduty_routing_key.clone(),
@@ -488,6 +496,7 @@ impl ForesterConfig {
             lookup_table_address: None,
             min_queue_items: None,
             enable_v1_multi_nullify: false,
+            enable_v1_presort: false,
             work_item_batch_size: 50,
         })
     }
@@ -511,6 +520,7 @@ impl Clone for ForesterConfig {
             lookup_table_address: self.lookup_table_address,
             min_queue_items: self.min_queue_items,
             enable_v1_multi_nullify: self.enable_v1_multi_nullify,
+            enable_v1_presort: self.enable_v1_presort,
             work_item_batch_size: self.work_item_batch_size,
         }
     }

forester/src/epoch_manager.rs

@@ -13,6 +13,7 @@ use dashmap::DashMap;
 use forester_utils::{
     forester_epoch::{get_epoch_phases, Epoch, ForesterSlot, TreeAccounts, TreeForesterSchedule},
     rpc_pool::SolanaRpcPool,
+    utils::wait_for_indexer,
 };
 use futures::future::join_all;
 use light_client::{
@@ -39,7 +40,7 @@ use solana_sdk::{
     transaction::TransactionError,
 };
 use tokio::{
-    sync::{mpsc, oneshot, Mutex},
+    sync::{mpsc, oneshot, Mutex, Semaphore},
     task::JoinHandle,
     time::{sleep, Instant, MissedTickBehavior},
 };
@@ -280,6 +281,8 @@ pub struct EpochManager<R: Rpc + Indexer> {
     run_id: Arc<str>,
     /// Per-epoch registration trackers to coordinate re-finalization when new foresters register mid-epoch
     registration_trackers: Arc<DashMap<u64, Arc<RegistrationTracker>>>,
+    /// Process-wide limiter for V1 transaction sends across all trees.
+    v1_send_permits: Arc<Semaphore>,
 }
 
 impl<R: Rpc + Indexer> Clone for EpochManager<R> {
@@ -310,6 +313,7 @@ impl<R: Rpc + Indexer> Clone for EpochManager<R> {
             heartbeat: self.heartbeat.clone(),
             run_id: self.run_id.clone(),
             registration_trackers: self.registration_trackers.clone(),
+            v1_send_permits: self.v1_send_permits.clone(),
         }
     }
 }
@@ -333,6 +337,7 @@ impl<R: Rpc + Indexer> EpochManager<R> {
         run_id: String,
     ) -> Result<Self> {
         let authority = Arc::new(config.payer_keypair.insecure_clone());
+        let v1_send_permit_count = config.transaction_config.max_concurrent_sends.max(1);
         Ok(Self {
             config,
             protocol_config,
@@ -359,6 +364,7 @@ impl<R: Rpc + Indexer> EpochManager<R> {
             heartbeat,
             run_id: Arc::<str>::from(run_id),
             registration_trackers: Arc::new(DashMap::new()),
+            v1_send_permits: Arc::new(Semaphore::new(v1_send_permit_count)),
         })
     }
 
@@ -2242,6 +2248,19 @@ impl<R: Rpc + Indexer> EpochManager<R> {
         )
         .await?;
 
+        if let Err(error) = wait_for_indexer(&*rpc).await {
+            if should_emit_rate_limited_warning("v2_wait_for_indexer", Duration::from_secs(30)) {
+                warn!(
+                    event = "v2_wait_for_indexer_error",
+                    run_id = %self.run_id,
+                    tree = %tree_pubkey,
+                    error = %error,
+                    "Skipping V2 proof work because the indexer is not fresh"
+                );
+            }
+            return Ok(());
+        }
+
         // Try to send any cached proofs first
         let cached_send_start = Instant::now();
         if let Some(items_sent) = self
@@ -2265,8 +2284,14 @@ impl<R: Rpc + Indexer> EpochManager<R> {
 
         let mut estimated_slot = self.slot_tracker.estimated_current_slot();
 
-        // Polling interval for checking queue
-        const POLL_INTERVAL: Duration = Duration::from_millis(200);
+        // Adaptive queue polling: start responsive, then back off (capped) while the
+        // queue has nothing ready to process, and reset to the minimum as soon as work
+        // is found. A fixed 200ms poll made idle V2 trees re-fetch the queue ~5x/sec for
+        // the whole eligible window, which is the dominant source of wasted RPC/indexer
+        // load (and can exhaust a shared RPC credit budget).
+        const POLL_INTERVAL_MIN: Duration = Duration::from_millis(200);
+        const POLL_INTERVAL_MAX: Duration = Duration::from_secs(10);
+        let mut poll_interval = POLL_INTERVAL_MIN;
 
         'inner_processing_loop: loop {
             if estimated_slot >= forester_slot_details.end_solana_slot {
@@ -2334,9 +2359,12 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                             processing_start_time.elapsed(),
                         )
                         .await;
+                        // Work found: stay responsive while the queue drains.
+                        poll_interval = POLL_INTERVAL_MIN;
                     } else {
-                        // No items to process, wait before polling again
-                        tokio::time::sleep(POLL_INTERVAL).await;
+                        // Nothing ready: wait, then back off (capped) to avoid hammering RPC.
+                        tokio::time::sleep(poll_interval).await;
+                        poll_interval = (poll_interval * 2).min(POLL_INTERVAL_MAX);
                     }
                 }
                 Err(e) => {
@@ -2350,7 +2378,8 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                         error = ?e,
                         "V2 processing failed for tree"
                     );
-                    tokio::time::sleep(POLL_INTERVAL).await;
+                    tokio::time::sleep(poll_interval).await;
+                    poll_interval = (poll_interval * 2).min(POLL_INTERVAL_MAX);
                 }
             }
 
@@ -3062,8 +3091,7 @@ impl<R: Rpc + Indexer> EpochManager<R> {
             } else {
                 None
             },
-            enable_presort: self.config.enable_v1_multi_nullify
-                && !self.address_lookup_tables.is_empty(),
+            enable_presort: self.config.enable_v1_presort,
             work_item_batch_size: self.config.work_item_batch_size,
         };
 
@@ -3083,6 +3111,7 @@ impl<R: Rpc + Indexer> EpochManager<R> {
             &batched_tx_config,
             *tree_accounts,
             transaction_builder,
+            self.v1_send_permits.clone(),
         )
         .await?;
 
@@ -3161,6 +3190,9 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                     .external_services
                     .prover_max_wait_time
                     .unwrap_or(Duration::from_secs(600)),
+                max_concurrent_jobs: self.config.external_services.prover_max_concurrent_jobs,
+                network: std::env::var("FORESTER_NETWORK")
+                    .unwrap_or_else(|_| "default".to_string()),
             }),
             ops_cache: self.ops_cache.clone(),
             epoch_phases: epoch_info.phases.clone(),
@@ -4703,6 +4735,7 @@ mod tests {
                 send_tx_rate_limit: None,
                 prover_polling_interval: None,
                 prover_max_wait_time: None,
+                prover_max_concurrent_jobs: 4,
                 fallback_rpc_url: None,
                 fallback_indexer_url: None,
             },
@@ -4730,6 +4763,7 @@ mod tests {
             lookup_table_address: None,
             min_queue_items: None,
             enable_v1_multi_nullify: false,
+            enable_v1_presort: false,
             work_item_batch_size: 50,
         }
     }

forester/src/processor/v1/helpers.rs

@@ -96,15 +96,16 @@ pub async fn fetch_proofs_and_create_instructions<R: Rpc>(
     };
 
     let rpc = pool.get_connection().await?;
-    if let Err(e) = wait_for_indexer(&*rpc).await {
+    wait_for_indexer(&*rpc).await.map_err(|e| {
         if should_emit_rate_limited_warning("v1_wait_for_indexer", Duration::from_secs(30)) {
             warn!(
                 event = "v1_wait_for_indexer_error",
                 error = %e,
-                "Indexer not fully caught up, but proceeding anyway"
+                "Skipping V1 proof work because the indexer is not fresh"
             );
         }
-    }
+        e
+    })?;
 
     let address_proofs = if let Some((merkle_tree, addresses)) = address_data {
         let total_addresses = addresses.len();