refactor: reduce clone pressure in compressible trackers

forester-utils/src/address_staging_tree.rs

@@ -121,7 +121,7 @@ impl AddressStagingTree {
         low_element_next_values: &[[u8; 32]],
         low_element_indices: &[u64],
         low_element_next_indices: &[u64],
-        low_element_proofs: &[Vec<[u8; 32]>],
+        low_element_proofs: &[[[u8; 32]; HEIGHT]],
         leaves_hashchain: [u8; 32],
         zkp_batch_size: usize,
         epoch: u64,
@@ -145,15 +145,12 @@ impl AddressStagingTree {
         let inputs = get_batch_address_append_circuit_inputs::<HEIGHT>(
             next_index,
             old_root,
-            low_element_values.to_vec(),
-            low_element_next_values.to_vec(),
-            low_element_indices.iter().map(|v| *v as usize).collect(),
-            low_element_next_indices
-                .iter()
-                .map(|v| *v as usize)
-                .collect(),
-            low_element_proofs.to_vec(),
-            addresses.to_vec(),
+            low_element_values,
+            low_element_next_values,
+            low_element_indices,
+            low_element_next_indices,
+            low_element_proofs,
+            addresses,
             &mut self.sparse_tree,
             leaves_hashchain,
             zkp_batch_size,

forester/src/compressible/ctoken/compressor.rs

@@ -30,7 +30,7 @@ use crate::{
 pub struct CTokenCompressor<R: Rpc + Indexer> {
     rpc_pool: Arc<SolanaRpcPool<R>>,
     tracker: Arc<CTokenAccountTracker>,
-    payer_keypair: Keypair,
+    payer_keypair: Arc<Keypair>,
     transaction_policy: TransactionPolicy,
 }
 
@@ -39,7 +39,7 @@ impl<R: Rpc + Indexer> Clone for CTokenCompressor<R> {
         Self {
             rpc_pool: Arc::clone(&self.rpc_pool),
             tracker: Arc::clone(&self.tracker),
-            payer_keypair: self.payer_keypair.insecure_clone(),
+            payer_keypair: Arc::clone(&self.payer_keypair),
             transaction_policy: self.transaction_policy,
         }
     }
@@ -55,7 +55,7 @@ impl<R: Rpc + Indexer> CTokenCompressor<R> {
         Self {
             rpc_pool,
             tracker,
-            payer_keypair,
+            payer_keypair: Arc::new(payer_keypair),
             transaction_policy,
         }
     }
@@ -253,7 +253,7 @@ impl<R: Rpc + Indexer> CTokenCompressor<R> {
         send_and_confirm_with_tracking(
             &mut *rpc,
             &[ix],
-            &self.payer_keypair,
+            self.payer_keypair.as_ref(),
             self.transaction_policy,
             &*self.tracker,
             &pubkeys,

forester/src/compressible/ctoken/state.rs

@@ -1,4 +1,4 @@
-use std::sync::atomic::AtomicU64;
+use std::sync::{atomic::AtomicU64, Arc};
 
 use borsh::BorshDeserialize;
 use dashmap::{DashMap, DashSet};
@@ -126,7 +126,7 @@ impl CTokenAccountTracker {
 
         let state = CTokenAccountState {
             pubkey,
-            account: ctoken,
+            account: Arc::new(ctoken),
             lamports,
             compressible_slot,
             is_ata,

forester/src/compressible/ctoken/types.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use light_token_interface::state::Token;
 use solana_sdk::pubkey::Pubkey;
 
@@ -6,7 +8,7 @@ use crate::compressible::traits::CompressibleState;
 #[derive(Clone, Debug)]
 pub struct CTokenAccountState {
     pub pubkey: Pubkey,
-    pub account: Token,
+    pub account: Arc<Token>,
     pub lamports: u64,
     /// Ready to compress when current_slot > compressible_slot
     pub compressible_slot: u64,

forester/src/compressible/mint/compressor.rs

@@ -30,7 +30,7 @@ use crate::{
 pub struct MintCompressor<R: Rpc + Indexer> {
     rpc_pool: Arc<SolanaRpcPool<R>>,
     tracker: Arc<MintAccountTracker>,
-    payer_keypair: Keypair,
+    payer_keypair: Arc<Keypair>,
     transaction_policy: TransactionPolicy,
 }
 
@@ -39,7 +39,7 @@ impl<R: Rpc + Indexer> Clone for MintCompressor<R> {
         Self {
             rpc_pool: Arc::clone(&self.rpc_pool),
             tracker: Arc::clone(&self.tracker),
-            payer_keypair: self.payer_keypair.insecure_clone(),
+            payer_keypair: Arc::clone(&self.payer_keypair),
             transaction_policy: self.transaction_policy,
         }
     }
@@ -55,7 +55,7 @@ impl<R: Rpc + Indexer> MintCompressor<R> {
         Self {
             rpc_pool,
             tracker,
-            payer_keypair,
+            payer_keypair: Arc::new(payer_keypair),
             transaction_policy,
         }
     }
@@ -117,7 +117,7 @@ impl<R: Rpc + Indexer> MintCompressor<R> {
         send_and_confirm_with_tracking(
             &mut *rpc,
             &instructions,
-            &self.payer_keypair,
+            self.payer_keypair.as_ref(),
             self.transaction_policy,
             &*self.tracker,
             &pubkeys,
@@ -160,26 +160,26 @@ impl<R: Rpc + Indexer> MintCompressor<R> {
         self.tracker.mark_pending(&all_pubkeys);
 
         // Create futures for each mint
-        let compression_futures = mint_states.iter().cloned().map(|mint_state| {
+        let compression_futures = mint_states.iter().map(|mint_state| {
             let compressor = self.clone();
             let cancelled = cancelled.clone();
             async move {
                 // Check cancellation before processing
                 if cancelled.load(Ordering::Relaxed) {
                     compressor.tracker.unmark_pending(&[mint_state.pubkey]);
                     return CompressionOutcome::Failed {
-                        state: mint_state,
+                        state: mint_state.clone(),
                         error: CompressionTaskError::Cancelled,
                     };
                 }
 
-                match compressor.compress(&mint_state).await {
+                match compressor.compress(mint_state).await {
                     Ok(sig) => CompressionOutcome::Compressed {
                         signature: sig,
-                        state: mint_state,
+                        state: mint_state.clone(),
                     },
                     Err(e) => CompressionOutcome::Failed {
-                        state: mint_state,
+                        state: mint_state.clone(),
                         error: e.into(),
                     },
                 }
@@ -259,7 +259,7 @@ impl<R: Rpc + Indexer> MintCompressor<R> {
         let signature = send_and_confirm_with_tracking(
             &mut *rpc,
             &[ix],
-            &self.payer_keypair,
+            self.payer_keypair.as_ref(),
             self.transaction_policy,
             &*self.tracker,
             &tracked_pubkeys,

forester/src/compressible/mint/state.rs

@@ -1,4 +1,4 @@
-use std::sync::atomic::AtomicU64;
+use std::sync::{atomic::AtomicU64, Arc};
 
 use borsh::BorshDeserialize;
 use dashmap::{DashMap, DashSet};
@@ -109,7 +109,7 @@ impl MintAccountTracker {
             pubkey,
             mint_seed,
             compressed_address,
-            mint,
+            mint: Arc::new(mint),
             lamports,
             compressible_slot,
         };

forester/src/compressible/mint/types.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use light_token_interface::state::Mint;
 use solana_sdk::pubkey::Pubkey;
 
@@ -8,7 +10,7 @@ pub struct MintAccountState {
     pub pubkey: Pubkey,
     pub mint_seed: Pubkey,
     pub compressed_address: [u8; 32],
-    pub mint: Mint,
+    pub mint: Arc<Mint>,
     pub lamports: u64,
     /// Ready to compress when current_slot > compressible_slot
     pub compressible_slot: u64,

forester/src/compressible/pda/compressor.rs

@@ -55,7 +55,7 @@ pub struct CachedProgramConfig {
 pub struct PdaCompressor<R: Rpc + Indexer> {
     rpc_pool: Arc<SolanaRpcPool<R>>,
     tracker: Arc<PdaAccountTracker>,
-    payer_keypair: Keypair,
+    payer_keypair: Arc<Keypair>,
     transaction_policy: TransactionPolicy,
 }
 
@@ -64,7 +64,7 @@ impl<R: Rpc + Indexer> Clone for PdaCompressor<R> {
         Self {
             rpc_pool: Arc::clone(&self.rpc_pool),
             tracker: Arc::clone(&self.tracker),
-            payer_keypair: self.payer_keypair.insecure_clone(),
+            payer_keypair: Arc::clone(&self.payer_keypair),
             transaction_policy: self.transaction_policy,
         }
     }
@@ -80,7 +80,7 @@ impl<R: Rpc + Indexer> PdaCompressor<R> {
         Self {
             rpc_pool,
             tracker,
-            payer_keypair,
+            payer_keypair: Arc::new(payer_keypair),
             transaction_policy,
         }
     }
@@ -171,10 +171,12 @@ impl<R: Rpc + Indexer> PdaCompressor<R> {
         self.tracker.mark_pending(&all_pubkeys);
 
         // Create futures for each account
-        let compression_futures = account_states.iter().cloned().map(|account_state| {
+        let program_config = Arc::new(program_config.clone());
+        let cached_config = Arc::new(cached_config.clone());
+        let compression_futures = account_states.iter().map(|account_state| {
             let compressor = self.clone();
-            let program_config = program_config.clone();
-            let cached_config = cached_config.clone();
+            let program_config = Arc::clone(&program_config);
+            let cached_config = Arc::clone(&cached_config);
             let cancelled = cancelled.clone();
 
             async move {
@@ -183,21 +185,21 @@ impl<R: Rpc + Indexer> PdaCompressor<R> {
                     // Unmark since we won't process this account
                     compressor.tracker.unmark_pending(&[account_state.pubkey]);
                     return CompressionOutcome::Failed {
-                        state: account_state,
+                        state: account_state.clone(),
                         error: CompressionTaskError::Cancelled,
                     };
                 }
 
                 match compressor
-                    .compress(&account_state, &program_config, &cached_config)
+                    .compress(account_state, &program_config, &cached_config)
                     .await
                 {
                     Ok(sig) => CompressionOutcome::Compressed {
                         signature: sig,
-                        state: account_state,
+                        state: account_state.clone(),
                     },
                     Err(e) => CompressionOutcome::Failed {
-                        state: account_state,
+                        state: account_state.clone(),
                         error: e.into(),
                     },
                 }
@@ -317,7 +319,7 @@ impl<R: Rpc + Indexer> PdaCompressor<R> {
         send_and_confirm_with_tracking(
             &mut *rpc,
             &[ix],
-            &self.payer_keypair,
+            self.payer_keypair.as_ref(),
             self.transaction_policy,
             &*self.tracker,
             &pubkeys,
@@ -396,7 +398,7 @@ impl<R: Rpc + Indexer> PdaCompressor<R> {
         );
 
         let payer_pubkey = self.payer_keypair.pubkey();
-        let signers = [&self.payer_keypair];
+        let signers = [self.payer_keypair.as_ref()];
         let instructions = vec![ix];
         let priority_fee_accounts = collect_priority_fee_accounts(payer_pubkey, &instructions);
         let signature = send_transaction_with_policy(