feat: disable v1 state multi-nullify when queue exceeds 10,000 items

Adds queue_item_count to BuildTransactionBatchConfig and disables
multi-nullify when the queue is too large, falling back to single
nullify for more reliable throughput. Renames use_dedup to
use_multi_nullify for consistency.

Author: ananas-block

forester/src/cli.rs

@@ -295,6 +295,14 @@ pub struct StartArgs {
     )]
     pub min_queue_items: Option<usize>,
 
+    #[arg(
+        long,
+        env = "ENABLE_V1_MULTI_NULLIFY",
+        help = "Enable nullify_state_v1_multi instruction for batching 2-4 V1 state nullifications per instruction. Requires --lookup-table-address.",
+        default_value = "false"
+    )]
+    pub enable_v1_multi_nullify: bool,
+
     #[arg(
         long,
         env = "API_SERVER_PORT",

forester/src/config.rs

@@ -33,6 +33,9 @@ pub struct ForesterConfig {
     /// Minimum queue items before processing V1 state nullifications.
     /// Delays processing to allow dedup grouping. Only applies when lookup_table_address is set.
     pub min_queue_items: Option<usize>,
+    /// 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,
 }
 
 #[derive(Debug, Clone)]
@@ -425,6 +428,7 @@ impl ForesterConfig {
                 })
                 .transpose()?,
             min_queue_items: args.min_queue_items,
+            enable_v1_multi_nullify: args.enable_v1_multi_nullify,
         })
     }
 
@@ -480,6 +484,7 @@ impl ForesterConfig {
             compressible_config: None,
             lookup_table_address: None,
             min_queue_items: None,
+            enable_v1_multi_nullify: false,
         })
     }
 }
@@ -501,6 +506,7 @@ impl Clone for ForesterConfig {
             compressible_config: self.compressible_config.clone(),
             lookup_table_address: self.lookup_table_address,
             min_queue_items: self.min_queue_items,
+            enable_v1_multi_nullify: self.enable_v1_multi_nullify,
         }
     }
 }

forester/src/epoch_manager.rs

@@ -275,7 +275,7 @@ pub struct EpochManager<R: Rpc + Indexer> {
     mint_tracker: Option<Arc<crate::compressible::mint::MintAccountTracker>>,
     /// Cached zkp_batch_size per tree to filter queue updates below threshold
     zkp_batch_sizes: Arc<DashMap<Pubkey, u64>>,
-    address_lookup_tables: Arc<tokio::sync::RwLock<Vec<AddressLookupTableAccount>>>,
+    address_lookup_tables: Arc<Vec<AddressLookupTableAccount>>,
     heartbeat: Arc<ServiceHeartbeat>,
     run_id: Arc<str>,
     /// Per-epoch registration trackers to coordinate re-finalization when new foresters register mid-epoch
@@ -328,7 +328,7 @@ impl<R: Rpc + Indexer> EpochManager<R> {
         compressible_tracker: Option<Arc<CTokenAccountTracker>>,
         pda_tracker: Option<Arc<crate::compressible::pda::PdaAccountTracker>>,
         mint_tracker: Option<Arc<crate::compressible::mint::MintAccountTracker>>,
-        address_lookup_tables: Arc<tokio::sync::RwLock<Vec<AddressLookupTableAccount>>>,
+        address_lookup_tables: Arc<Vec<AddressLookupTableAccount>>,
         heartbeat: Arc<ServiceHeartbeat>,
         run_id: String,
     ) -> Result<Self> {
@@ -1096,21 +1096,6 @@ impl<R: Rpc + Indexer> EpochManager<R> {
         debug!("Recovered registration info for epoch {}", epoch);
         update_epoch_registered(epoch);
 
-        // Extend ALT with new forester epoch PDA if ALT is configured
-        let forester_epoch_pda_pubkey =
-            get_forester_epoch_pda_from_authority(&self.config.derivation_pubkey, epoch).0;
-        if let Err(e) = self
-            .extend_alt_with_forester_pda(forester_epoch_pda_pubkey)
-            .await
-        {
-            warn!(
-                event = "extend_alt_failed",
-                epoch,
-                error = ?e,
-                "Failed to extend ALT with forester PDA, continuing with static account"
-            );
-        }
-
         // Wait for the active phase
         registration_info = match self.wait_for_active_phase(&registration_info).await? {
             Some(info) => info,
@@ -1395,51 +1380,6 @@ impl<R: Rpc + Indexer> EpochManager<R> {
         }
     }
 
-    async fn extend_alt_with_forester_pda(&self, forester_epoch_pda: Pubkey) -> anyhow::Result<()> {
-        let alt_address = match self.config.lookup_table_address {
-            Some(addr) => addr,
-            None => return Ok(()),
-        };
-
-        // Check if the PDA is already in the ALT
-        {
-            let alt = self.address_lookup_tables.read().await;
-            if alt
-                .iter()
-                .any(|t| t.addresses.contains(&forester_epoch_pda))
-            {
-                return Ok(());
-            }
-        }
-
-        let extend_ix = light_client::rpc::lut::instruction::extend_lookup_table(
-            alt_address,
-            self.config.payer_keypair.pubkey(),
-            Some(self.config.payer_keypair.pubkey()),
-            vec![forester_epoch_pda],
-        );
-        let payer_pubkey = self.config.payer_keypair.pubkey();
-        let mut rpc = self.rpc_pool.get_connection().await?;
-        rpc.create_and_send_transaction(&[extend_ix], &payer_pubkey, &[&self.config.payer_keypair])
-            .await
-            .map_err(|e| anyhow::anyhow!("Failed to extend ALT: {e}"))?;
-
-        // Reload the ALT from on-chain
-        let updated_lut = load_lookup_table_async(&*rpc, alt_address).await?;
-        info!(
-            event = "alt_extended",
-            lookup_table = %alt_address,
-            new_address = %forester_epoch_pda,
-            address_count = updated_lut.addresses.len(),
-            "Extended ALT with forester epoch PDA"
-        );
-
-        let mut alt = self.address_lookup_tables.write().await;
-        *alt = vec![updated_lut];
-
-        Ok(())
-    }
-
     async fn recover_registration_info_internal(
         &self,
         epoch: u64,
@@ -1574,14 +1514,13 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                 };
                 let payer = self.config.payer_keypair.pubkey();
                 let signers = [&self.config.payer_keypair];
-                let alt_guard = self.address_lookup_tables.read().await;
                 send_smart_transaction(
                     &mut *rpc,
                     SendSmartTransactionConfig {
                         instructions: vec![ix],
                         payer: &payer,
                         signers: &signers,
-                        address_lookup_tables: &alt_guard,
+                        address_lookup_tables: &self.address_lookup_tables,
                         compute_budget: ComputeBudgetConfig {
                             compute_unit_price: priority_fee,
                             compute_unit_limit: Some(self.config.transaction_config.cu_limit),
@@ -1986,14 +1925,13 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                 };
                 let payer = self.config.payer_keypair.pubkey();
                 let signers = [&self.config.payer_keypair];
-                let alt_guard = self.address_lookup_tables.read().await;
                 match send_smart_transaction(
                     &mut *rpc,
                     SendSmartTransactionConfig {
                         instructions: vec![ix],
                         payer: &payer,
                         signers: &signers,
-                        address_lookup_tables: &alt_guard,
+                        address_lookup_tables: &self.address_lookup_tables,
                         compute_budget: ComputeBudgetConfig {
                             compute_unit_price: priority_fee,
                             compute_unit_limit: Some(self.config.transaction_config.cu_limit),
@@ -3055,6 +2993,7 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                 compute_unit_limit: Some(self.config.transaction_config.cu_limit),
                 enable_priority_fees: self.config.transaction_config.enable_priority_fees,
                 max_concurrent_sends: Some(self.config.transaction_config.max_concurrent_sends),
+                queue_item_count: 0,
             },
             queue_config: self.config.queue_config,
             retry_config: RetryConfig {
@@ -3070,15 +3009,16 @@ impl<R: Rpc + Indexer> EpochManager<R> {
             min_queue_items: None, // set below after reading ALT
         };
 
-        let alt_snapshot = self.address_lookup_tables.read().await.clone();
-        if !alt_snapshot.is_empty() {
+        let alt_snapshot = (*self.address_lookup_tables).clone();
+        if self.config.enable_v1_multi_nullify && !alt_snapshot.is_empty() {
             batched_tx_config.min_queue_items = self.config.min_queue_items;
         }
         let transaction_builder = Arc::new(EpochManagerTransactions::new(
             self.rpc_pool.clone(),
             epoch_info.epoch,
             self.tx_cache.clone(),
             alt_snapshot,
+            self.config.enable_v1_multi_nullify,
         ));
 
         let num_sent = send_batched_transactions(
@@ -3280,14 +3220,13 @@ impl<R: Rpc + Indexer> EpochManager<R> {
         }
 
         // No existing processor - create new one
-        let alt_guard = self.address_lookup_tables.read().await;
         let batch_context = self.build_batch_context(
             epoch_info,
             tree_accounts,
             None,
             None,
             None,
-            Arc::new(alt_guard.clone()),
+            self.address_lookup_tables.clone(),
         );
         let processor = Arc::new(Mutex::new(
             QueueProcessor::new(batch_context, StateTreeStrategy).await?,
@@ -3342,14 +3281,13 @@ impl<R: Rpc + Indexer> EpochManager<R> {
         }
 
         // No existing processor - create new one
-        let alt_guard = self.address_lookup_tables.read().await;
         let batch_context = self.build_batch_context(
             epoch_info,
             tree_accounts,
             None,
             None,
             None,
-            Arc::new(alt_guard.clone()),
+            self.address_lookup_tables.clone(),
         );
         let processor = Arc::new(Mutex::new(
             QueueProcessor::new(batch_context, AddressTreeStrategy).await?,
@@ -3927,14 +3865,13 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                 let instruction_count = instructions.len();
                 let payer = self.config.payer_keypair.pubkey();
                 let signers = [&self.config.payer_keypair];
-                let alt_guard = self.address_lookup_tables.read().await;
                 match send_smart_transaction(
                     &mut *rpc,
                     SendSmartTransactionConfig {
                         instructions,
                         payer: &payer,
                         signers: &signers,
-                        address_lookup_tables: &alt_guard,
+                        address_lookup_tables: &self.address_lookup_tables,
                         compute_budget: ComputeBudgetConfig {
                             compute_unit_price: priority_fee,
                             compute_unit_limit: Some(self.config.transaction_config.cu_limit),
@@ -4103,14 +4040,13 @@ impl<R: Rpc + Indexer> EpochManager<R> {
             .await?;
         let payer = self.config.payer_keypair.pubkey();
         let signers = [&self.config.payer_keypair];
-        let alt_guard = self.address_lookup_tables.read().await;
         match send_smart_transaction(
             &mut rpc,
             SendSmartTransactionConfig {
                 instructions: vec![ix],
                 payer: &payer,
                 signers: &signers,
-                address_lookup_tables: &alt_guard,
+                address_lookup_tables: &self.address_lookup_tables,
                 compute_budget: ComputeBudgetConfig {
                     compute_unit_price: priority_fee,
                     compute_unit_limit: Some(self.config.transaction_config.cu_limit),
@@ -4546,28 +4482,28 @@ pub async fn run_service<R: Rpc + Indexer>(
                 let address_lookup_tables = {
                     if let Some(lut_address) = config.lookup_table_address {
                         let rpc = rpc_pool.get_connection().await?;
-                        match load_lookup_table_async(&*rpc, lut_address).await {
-                            Ok(lut) => {
-                                info!(
-                                    event = "lookup_table_loaded",
+                        let lut = load_lookup_table_async(&*rpc, lut_address).await
+                            .map_err(|e| {
+                                error!(
+                                    event = "lookup_table_load_failed",
                                     run_id = %run_id_for_logs,
                                     lookup_table = %lut_address,
-                                    address_count = lut.addresses.len(),
-                                    "Loaded lookup table"
+                                    error = %e,
+                                    "Failed to load lookup table"
                                 );
-                                Arc::new(tokio::sync::RwLock::new(vec![lut]))
-                            }
-                            Err(e) => {
-                                debug!(
-                                    "Lookup table {} not available: {}. Using legacy transactions.",
-                                    lut_address, e
-                                );
-                                Arc::new(tokio::sync::RwLock::new(Vec::new()))
-                            }
-                        }
+                                e
+                            })?;
+                        info!(
+                            event = "lookup_table_loaded",
+                            run_id = %run_id_for_logs,
+                            lookup_table = %lut_address,
+                            address_count = lut.addresses.len(),
+                            "Loaded lookup table"
+                        );
+                        Arc::new(vec![lut])
                     } else {
-                        debug!("No lookup table address configured. Using legacy transactions.");
-                        Arc::new(tokio::sync::RwLock::new(Vec::new()))
+                        debug!("No lookup table address configured. Using v1 state single nullify transactions.");
+                        Arc::new(Vec::new())
                     }
                 };
 
@@ -4738,6 +4674,7 @@ mod tests {
             compressible_config: None,
             lookup_table_address: None,
             min_queue_items: None,
+            enable_v1_multi_nullify: false,
         }
     }
 

forester/src/processor/v1/config.rs

@@ -32,4 +32,7 @@ pub struct BuildTransactionBatchConfig {
     pub compute_unit_limit: Option<u32>,
     pub enable_priority_fees: bool,
     pub max_concurrent_sends: Option<usize>,
+    /// Number of items in the queue at the time of batch preparation.
+    /// Used to disable multi-nullify when queue is very large (>10,000 items).
+    pub queue_item_count: usize,
 }