fix(forester): restore Clone + compile tests after tracker cleanup

The cleanup commit changed get_ready_to_compress to return Vec<Pubkey>
and CTokenCompressor/MintCompressor/PdaCompressor::new to take
Arc<Keypair>, but the compressible tests weren't updated. Added
get_ready_states / get_ready_states_for_program helpers returning
Vec<S> for test ergonomics, wrapped test keypair args in Arc, and
re-ran cargo +nightly fmt across the refactored modules.

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

Author: sergeytimoshin

forester/src/compressible/mint/compressor.rs

@@ -171,20 +171,24 @@ impl<R: Rpc + Indexer> MintCompressor<R> {
                     };
                 }
 
-                let mint_state =
-                    match compressor.tracker.accounts().get(&pubkey).map(|r| r.clone()) {
-                        Some(state) => state,
-                        None => {
-                            compressor.tracker.unmark_pending(&[pubkey]);
-                            return CompressionOutcome::Failed {
-                                pubkey,
-                                error: CompressionTaskError::Failed(anyhow::anyhow!(
-                                    "mint {} removed from tracker before compression",
-                                    pubkey
-                                )),
-                            };
-                        }
-                    };
+                let mint_state = match compressor
+                    .tracker
+                    .accounts()
+                    .get(&pubkey)
+                    .map(|r| r.clone())
+                {
+                    Some(state) => state,
+                    None => {
+                        compressor.tracker.unmark_pending(&[pubkey]);
+                        return CompressionOutcome::Failed {
+                            pubkey,
+                            error: CompressionTaskError::Failed(anyhow::anyhow!(
+                                "mint {} removed from tracker before compression",
+                                pubkey
+                            )),
+                        };
+                    }
+                };
 
                 match compressor.compress(&mint_state).await {
                     Ok(sig) => CompressionOutcome::Compressed {

forester/src/compressible/pda/compressor.rs

@@ -187,20 +187,24 @@ impl<R: Rpc + Indexer> PdaCompressor<R> {
                 }
 
                 // Look up account state from tracker; it may have been removed
-                let account_state =
-                    match compressor.tracker.accounts().get(&pubkey).map(|r| r.clone()) {
-                        Some(state) => state,
-                        None => {
-                            compressor.tracker.unmark_pending(&[pubkey]);
-                            return CompressionOutcome::Failed {
-                                pubkey,
-                                error: CompressionTaskError::Failed(anyhow::anyhow!(
-                                    "account {} removed from tracker before compression",
-                                    pubkey
-                                )),
-                            };
-                        }
-                    };
+                let account_state = match compressor
+                    .tracker
+                    .accounts()
+                    .get(&pubkey)
+                    .map(|r| r.clone())
+                {
+                    Some(state) => state,
+                    None => {
+                        compressor.tracker.unmark_pending(&[pubkey]);
+                        return CompressionOutcome::Failed {
+                            pubkey,
+                            error: CompressionTaskError::Failed(anyhow::anyhow!(
+                                "account {} removed from tracker before compression",
+                                pubkey
+                            )),
+                        };
+                    }
+                };
 
                 match compressor
                     .compress(&account_state, &program_config, &cached_config)

forester/src/compressible/pda/state.rs

@@ -85,6 +85,23 @@ impl PdaAccountTracker {
             .collect()
     }
 
+    pub fn get_ready_states_for_program(
+        &self,
+        program_id: &Pubkey,
+        current_slot: u64,
+    ) -> Vec<PdaAccountState> {
+        let pending = self.pending();
+        self.accounts()
+            .iter()
+            .filter(|entry| {
+                entry.value().program_id == *program_id
+                    && entry.value().is_ready_to_compress(current_slot)
+                    && !pending.contains(entry.key())
+            })
+            .map(|entry| entry.value().clone())
+            .collect()
+    }
+
     pub fn update_from_account(
         &self,
         pubkey: Pubkey,

forester/src/compressible/traits.rs

@@ -60,7 +60,7 @@ pub enum CompressionOutcome {
 
 pub type CompressionOutcomes = Vec<CompressionOutcome>;
 
-pub trait CompressibleState: Send + Sync {
+pub trait CompressibleState: Clone + Send + Sync {
     fn pubkey(&self) -> &Pubkey;
     fn lamports(&self) -> u64;
     fn compressible_slot(&self) -> u64;
@@ -138,6 +138,19 @@ pub trait CompressibleTracker<S: CompressibleState>: Send + Sync {
             .map(|entry| *entry.key())
             .collect()
     }
+
+    /// Clone of ready-to-compress states. Prefer `get_ready_to_compress` in
+    /// production; this helper exists for tests that need full state fields.
+    fn get_ready_states(&self, current_slot: u64) -> Vec<S> {
+        let pending = self.pending();
+        self.accounts()
+            .iter()
+            .filter(|entry| {
+                entry.value().is_ready_to_compress(current_slot) && !pending.contains(entry.key())
+            })
+            .map(|entry| entry.value().clone())
+            .collect()
+    }
 }
 
 /// Allows AccountSubscriber to work with any tracker type.

forester/src/epoch_manager/compression.rs

@@ -12,18 +12,14 @@ use light_registry::ForesterEpochPda;
 use solana_program::pubkey::Pubkey;
 use tracing::{debug, error, info, trace, warn};
 
-use crate::{
-    compressible::{
-        traits::{
-            Cancelled, CompressibleState, CompressibleTracker, CompressionOutcome,
-            CompressionTaskError,
-        },
-        CTokenCompressor, CompressibleConfig,
+use super::EpochManager;
+use crate::compressible::{
+    traits::{
+        Cancelled, CompressibleState, CompressibleTracker, CompressionOutcome, CompressionTaskError,
     },
+    CTokenCompressor, CompressibleConfig,
 };
 
-use super::EpochManager;
-
 impl<R: Rpc + Indexer> EpochManager<R> {
     pub(crate) async fn dispatch_compression(
         &self,

forester/src/epoch_manager/context.rs

@@ -1,24 +1,23 @@
-use std::{sync::Arc, time::Duration};
+use std::{
+    sync::{atomic::AtomicU64, Arc},
+    time::Duration,
+};
 
-use light_client::{
-    indexer::Indexer,
-    rpc::Rpc,
+use forester_utils::{
+    forester_epoch::{Epoch, TreeAccounts},
+    rpc_pool::SolanaRpcPool,
 };
+use light_client::{indexer::Indexer, rpc::Rpc};
 use light_compressed_account::TreeType;
-use light_registry::utils::get_forester_epoch_pda_from_authority;
+use light_registry::{
+    protocol_config::state::ProtocolConfig, utils::get_forester_epoch_pda_from_authority,
+};
 use solana_sdk::{
     address_lookup_table::AddressLookupTableAccount,
     signature::{Keypair, Signer},
 };
 use tokio::sync::Mutex;
 
-use forester_utils::{
-    forester_epoch::{Epoch, TreeAccounts},
-    rpc_pool::SolanaRpcPool,
-};
-use light_registry::protocol_config::state::ProtocolConfig;
-use std::sync::atomic::AtomicU64;
-
 use crate::{
     logging::ServiceHeartbeat,
     priority_fee::PriorityFeeConfig,

forester/src/epoch_manager/mod.rs

@@ -1,3 +1,4 @@
+mod compression;
 pub(crate) mod context;
 mod monitor;
 mod pipeline;
@@ -7,53 +8,37 @@ mod reporting;
 pub(crate) mod tracker;
 mod v1;
 mod v2;
-mod compression;
 
 use std::{
     sync::Arc,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
 use anyhow::anyhow;
-use light_client::{
-    indexer::Indexer,
-    rpc::Rpc,
-};
+use forester_utils::{forester_epoch::TreeAccounts, rpc_pool::SolanaRpcPool};
+use light_client::{indexer::Indexer, rpc::Rpc};
 use light_compressed_account::TreeType;
-use forester_utils::{
-    forester_epoch::TreeAccounts,
-    rpc_pool::SolanaRpcPool,
-};
 use light_registry::protocol_config::state::ProtocolConfig;
-use solana_sdk::{
-    address_lookup_table::AddressLookupTableAccount,
-    signature::Signer,
-};
+use solana_sdk::{address_lookup_table::AddressLookupTableAccount, signature::Signer};
 use tokio::{
     sync::{mpsc, oneshot, watch, Mutex},
     task::JoinHandle,
     time::{sleep, Instant, MissedTickBehavior},
 };
 use tracing::{debug, error, info, info_span};
 
+use self::{context::ForesterContext, processor_pool::ProcessorPool, tracker::EpochTracker};
 use crate::{
     compressible::CTokenAccountTracker,
     errors::InitializationError,
     logging::ServiceHeartbeat,
     processor::tx_cache::ProcessedHashCache,
+    queue_helpers::QueueItemData,
     slot_tracker::SlotTracker,
     tree_data_sync::{fetch_protocol_group_authority, fetch_trees},
     ForesterConfig, Result,
 };
 
-use self::{
-    context::ForesterContext,
-    processor_pool::ProcessorPool,
-    tracker::EpochTracker,
-};
-
-use crate::queue_helpers::QueueItemData;
-
 // ── Public re-exports (preserve existing public API) ─────────────────────
 
 /// Timing for a single circuit type (circuit inputs + proof generation)
@@ -438,7 +423,9 @@ fn spawn_heartbeat_task(
             let delta_queues_finished = current
                 .queues_finished
                 .saturating_sub(previous.queues_finished);
-            let delta_items = current.items_processed.saturating_sub(previous.items_processed);
+            let delta_items = current
+                .items_processed
+                .saturating_sub(previous.items_processed);
             let delta_work_reports = current
                 .work_reports_sent
                 .saturating_sub(previous.work_reports_sent);
@@ -1005,7 +992,10 @@ mod tests {
         };
         let work_item = WorkItem {
             tree_account,
-            queue_item_data: QueueItemData { hash: [0u8; 32], index: 0 },
+            queue_item_data: QueueItemData {
+                hash: [0u8; 32],
+                index: 0,
+            },
         };
         assert!(work_item.is_address_tree());
         assert!(!work_item.is_state_tree());
@@ -1022,7 +1012,10 @@ mod tests {
         };
         let work_item = WorkItem {
             tree_account,
-            queue_item_data: QueueItemData { hash: [0u8; 32], index: 0 },
+            queue_item_data: QueueItemData {
+                hash: [0u8; 32],
+                index: 0,
+            },
         };
         assert!(!work_item.is_address_tree());
         assert!(work_item.is_state_tree());

forester/src/epoch_manager/monitor.rs

@@ -3,26 +3,20 @@
 use std::{sync::Arc, time::Duration};
 
 use anyhow::anyhow;
-use forester_utils::forester_epoch::{
-    get_epoch_phases, TreeAccounts, TreeForesterSchedule,
-};
+use forester_utils::forester_epoch::{get_epoch_phases, TreeAccounts, TreeForesterSchedule};
 use light_client::{indexer::Indexer, rpc::Rpc};
 use solana_program::{native_token::LAMPORTS_PER_SOL, pubkey::Pubkey};
 use solana_sdk::signature::Signer;
 use tokio::sync::mpsc;
 use tracing::{debug, error, info, warn};
 
+use super::{context::should_skip_tree, EpochManager};
 use crate::{
     metrics::update_forester_sol_balance,
     slot_tracker::wait_until_slot_reached,
     tree_data_sync::{fetch_protocol_group_authority, fetch_trees},
 };
 
-use super::{
-    context::should_skip_tree,
-    EpochManager,
-};
-
 impl<R: Rpc + Indexer> EpochManager<R> {
     pub(super) async fn check_sol_balance_periodically(self: Arc<Self>) -> crate::Result<()> {
         let interval_duration = Duration::from_secs(300);
@@ -31,7 +25,10 @@ impl<R: Rpc + Indexer> EpochManager<R> {
         loop {
             interval.tick().await;
             match self.ctx.rpc_pool.get_connection().await {
-                Ok(rpc) => match rpc.get_balance(&self.ctx.config.payer_keypair.pubkey()).await {
+                Ok(rpc) => match rpc
+                    .get_balance(&self.ctx.config.payer_keypair.pubkey())
+                    .await
+                {
                     Ok(balance) => {
                         let balance_in_sol = balance as f64 / (LAMPORTS_PER_SOL as f64);
                         update_forester_sol_balance(
@@ -63,7 +60,11 @@ impl<R: Rpc + Indexer> EpochManager<R> {
     }
 
     pub(super) async fn discover_trees_periodically(self: Arc<Self>) -> crate::Result<()> {
-        let interval_secs = self.ctx.config.general_config.tree_discovery_interval_seconds;
+        let interval_secs = self
+            .ctx
+            .config
+            .general_config
+            .tree_discovery_interval_seconds;
         if interval_secs == 0 {
             info!(event = "tree_discovery_disabled", run_id = %self.ctx.run_id, "Tree discovery disabled (interval=0)");
             return Ok(());
@@ -429,7 +430,8 @@ impl<R: Rpc + Indexer> EpochManager<R> {
                     );
 
                     if let Err(e) =
-                        wait_until_slot_reached(&mut *rpc, &self.ctx.slot_tracker, wait_target).await
+                        wait_until_slot_reached(&mut *rpc, &self.ctx.slot_tracker, wait_target)
+                            .await
                     {
                         error!(
                             event = "epoch_monitor_wait_for_registration_failed",

forester/src/epoch_manager/pipeline.rs

@@ -7,14 +7,12 @@ use forester_utils::forester_epoch::{
 };
 use light_client::{indexer::Indexer, rpc::Rpc};
 use light_compressed_account::TreeType;
-use light_registry::{
-    protocol_config::state::EpochState,
-    ForesterEpochPda,
-};
+use light_registry::{protocol_config::state::EpochState, ForesterEpochPda};
 use solana_sdk::signature::Signer;
 use tokio::time::Instant;
 use tracing::{debug, error, info, instrument, trace, warn};
 
+use super::{context::should_skip_tree, tracker::RegistrationTracker, EpochManager};
 use crate::{
     errors::ForesterError,
     logging::should_emit_rate_limited_warning,
@@ -23,12 +21,6 @@ use crate::{
     ForesterEpochInfo,
 };
 
-use super::{
-    context::should_skip_tree,
-    tracker::RegistrationTracker,
-    EpochManager,
-};
-
 impl<R: Rpc + Indexer> EpochManager<R> {
     #[instrument(level = "debug", skip(self), fields(forester = %self.ctx.config.payer_keypair.pubkey(), epoch = epoch))]
     pub(super) async fn process_epoch(self: Arc<Self>, epoch: u64) -> crate::Result<()> {