refactor: forester: add default tree / epoch

forester/Cargo.toml

@@ -62,6 +62,7 @@ num-bigint = { workspace = true }
 kameo = "0.19"
 once_cell = "1.21.3"
 async-channel = "2.3"
+solana-pubkey = { workspace = true }
 
 [dev-dependencies]
 serial_test = { workspace = true }

forester/src/compressible/compressor.rs

@@ -2,7 +2,8 @@ use std::{str::FromStr, sync::Arc};
 
 use anchor_lang::{InstructionData, ToAccountMetas};
 use forester_utils::rpc_pool::SolanaRpcPool;
-use light_client::rpc::Rpc;
+use light_client::{indexer::TreeInfo, rpc::Rpc};
+use light_compressed_account::TreeType;
 use light_compressible::config::CompressibleConfig;
 use light_ctoken_interface::CTOKEN_PROGRAM_ID;
 use light_ctoken_sdk::compressed_token::compress_and_close::CompressAndCloseAccounts as CTokenAccounts;
@@ -11,13 +12,14 @@ use light_registry::{
     instruction::CompressAndClose,
 };
 use light_sdk::instruction::PackedAccounts;
+use solana_pubkey::pubkey;
 use solana_sdk::{
     instruction::Instruction,
     pubkey::Pubkey,
     signature::{Keypair, Signature},
     signer::Signer,
 };
-use tracing::{debug, warn};
+use tracing::{debug, info};
 
 use super::{state::CompressibleAccountTracker, types::CompressibleAccountState};
 use crate::Result;
@@ -83,13 +85,23 @@ impl<R: Rpc> Compressor<R> {
 
         // Get output tree from RPC
         let mut rpc = self.rpc_pool.get_connection().await?;
-        rpc.get_latest_active_state_trees()
-            .await
-            .map_err(|e| anyhow::anyhow!("Failed to get state tree info: {}", e))?;
 
-        let output_tree_info = rpc
-            .get_random_state_tree_info()
-            .map_err(|e| anyhow::anyhow!("Failed to get state tree info: {}", e))?;
+        // FIXME: Use latest active state tree after updating lookup tables
+        // rpc.get_latest_active_state_trees()
+        //     .await
+        //     .map_err(|e| anyhow::anyhow!("Failed to get state tree info: {}", e))?;
+        // let output_tree_info = rpc
+        //     .get_random_state_tree_info()
+        //     .map_err(|e| anyhow::anyhow!("Failed to get state tree info: {}", e))?;
+
+        let output_tree_info = TreeInfo {
+            tree: pubkey!("bmt1LryLZUMmF7ZtqESaw7wifBXLfXHQYoE4GAmrahU"),
+            queue: pubkey!("oq1na8gojfdUhsfCpyjNt6h4JaDWtHf1yQj4koBWfto"),
+            cpi_context: Some(pubkey!("cpi15BoVPKgEPw5o8wc2T816GE7b378nMXnhH3Xbq4y")),
+            tree_type: TreeType::StateV2,
+            next_tree_info: None,
+        };
+
         let output_queue = output_tree_info
             .get_output_pubkey()
             .map_err(|e| anyhow::anyhow!("Failed to get output queue: {}", e))?;
@@ -196,16 +208,15 @@ impl<R: Rpc> Compressor<R> {
             accounts.len()
         );
 
-        // Collect pubkeys for sync before creating instruction
-        let pubkeys: Vec<_> = account_states.iter().map(|state| state.pubkey).collect();
-
         let ix = Instruction {
             program_id: registry_program_id,
             accounts,
             data: instruction.data(),
         };
 
         // Send transaction
+        // Note: Account removal from tracker is handled by LogSubscriber which parses
+        // the "compress_and_close:<pubkey>" logs emitted by the registry program
         let signature = rpc
             .create_and_send_transaction(
                 &[ix],
@@ -215,10 +226,11 @@ impl<R: Rpc> Compressor<R> {
             .await
             .map_err(|e| anyhow::anyhow!("Failed to send transaction: {}", e))?;
 
-        // Sync accounts to verify they're closed
-        if let Err(e) = self.tracker.sync_accounts(&*rpc, &pubkeys).await {
-            warn!("Failed to sync accounts after compression: {:?}. Tracker will update via subscriptions.", e);
-        }
+        info!(
+            "compress_and_close tx with ({:?}) accounts sent {}",
+            account_states.iter().map(|a| a.pubkey.to_string()),
+            signature
+        );
 
         Ok(signature)
     }

forester/src/compressible/config.rs

@@ -13,7 +13,7 @@ pub struct CompressibleConfig {
 }
 
 fn default_batch_size() -> usize {
-    10
+    5
 }
 
 fn default_max_concurrent_batches() -> usize {

forester/src/compressible/mod.rs

@@ -9,5 +9,5 @@ pub use bootstrap::bootstrap_compressible_accounts;
 pub use compressor::Compressor;
 pub use config::CompressibleConfig;
 pub use state::CompressibleAccountTracker;
-pub use subscriber::AccountSubscriber;
+pub use subscriber::{AccountSubscriber, LogSubscriber};
 pub use types::CompressibleAccountState;

forester/src/compressible/state.rs

@@ -156,86 +156,6 @@ impl CompressibleAccountTracker {
 
         Ok(())
     }
-
-    /// Query accounts and update tracker: remove non-existent accounts, update lamports for existing ones
-    pub async fn sync_accounts<R: light_client::rpc::Rpc>(
-        &self,
-        rpc: &R,
-        pubkeys: &[Pubkey],
-    ) -> Result<()> {
-        // Query all accounts at once using get_multiple_accounts
-        let accounts = rpc.get_multiple_accounts(pubkeys).await?;
-
-        for (pubkey, account_opt) in pubkeys.iter().zip(accounts.iter()) {
-            match account_opt {
-                Some(account) => {
-                    // Check if account is closed (lamports == 0)
-                    if account.lamports == 0 {
-                        self.remove(pubkey);
-                        debug!("Removed closed account {} (lamports == 0)", pubkey);
-                        continue;
-                    }
-
-                    // Re-deserialize account data to verify it's still valid
-                    let ctoken = match CToken::try_from_slice(&account.data) {
-                        Ok(ct) => ct,
-                        Err(e) => {
-                            self.remove(pubkey);
-                            debug!(
-                                "Removed account {} (deserialization failed: {:?})",
-                                pubkey, e
-                            );
-                            continue;
-                        }
-                    };
-
-                    // Verify Compressible extension still exists
-                    let has_compressible_ext = ctoken.extensions.as_ref().is_some_and(|exts| {
-                        exts.iter()
-                            .any(|ext| matches!(ext, ExtensionStruct::Compressible(_)))
-                    });
-
-                    if !has_compressible_ext {
-                        self.remove(pubkey);
-                        debug!(
-                            "Removed account {} (missing Compressible extension)",
-                            pubkey
-                        );
-                        continue;
-                    }
-
-                    // Account is valid - update state
-                    if let Some(mut state) = self.accounts.get_mut(pubkey) {
-                        match calculate_compressible_slot(&ctoken, account.lamports) {
-                            Ok(compressible_slot) => {
-                                state.account = ctoken;
-                                state.lamports = account.lamports;
-                                state.compressible_slot = compressible_slot;
-                                debug!(
-                                    "Updated account {}: lamports={}, compressible_slot={}",
-                                    pubkey, account.lamports, compressible_slot
-                                );
-                            }
-                            Err(e) => {
-                                warn!(
-                                    "Failed to calculate compressible slot for account {}: {}. Removing from tracker.",
-                                    pubkey, e
-                                );
-                                drop(state);
-                                self.remove(pubkey);
-                            }
-                        }
-                    }
-                }
-                None => {
-                    // Account doesn't exist - remove from tracker
-                    self.remove(pubkey);
-                    debug!("Removed non-existent account {}", pubkey);
-                }
-            }
-        }
-        Ok(())
-    }
 }
 
 impl Default for CompressibleAccountTracker {

forester/src/compressible/subscriber.rs

@@ -5,29 +5,38 @@ use light_ctoken_interface::{COMPRESSIBLE_TOKEN_ACCOUNT_SIZE, CTOKEN_PROGRAM_ID}
 use solana_account_decoder::UiAccountEncoding;
 use solana_client::{
     nonblocking::pubsub_client::PubsubClient,
-    rpc_config::{RpcAccountInfoConfig, RpcProgramAccountsConfig},
-    rpc_response::{Response as RpcResponse, RpcKeyedAccount},
+    rpc_config::{
+        RpcAccountInfoConfig, RpcProgramAccountsConfig, RpcTransactionLogsConfig,
+        RpcTransactionLogsFilter,
+    },
+    rpc_response::{Response as RpcResponse, RpcKeyedAccount, RpcLogsResponse},
 };
 use solana_rpc_client_api::filter::RpcFilterType;
 use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey};
-use tokio::sync::oneshot;
+use tokio::sync::broadcast;
 use tracing::{debug, error, info};
 
 use super::state::CompressibleAccountTracker;
 use crate::Result;
 
+/// Registry program ID for subscribing to compress_and_close logs
+const REGISTRY_PROGRAM_ID_STR: &str = "Lighton6oQpVkeewmo2mcPTQQp7kYHr4fWpAgJyEmDX";
+
+/// Log prefix emitted by registry program when closing accounts
+const COMPRESS_AND_CLOSE_LOG_PREFIX: &str = "compress_and_close:";
+
 /// Subscribes to account changes for all compressible CToken accounts
 pub struct AccountSubscriber {
     ws_url: String,
     tracker: Arc<CompressibleAccountTracker>,
-    shutdown_rx: oneshot::Receiver<()>,
+    shutdown_rx: broadcast::Receiver<()>,
 }
 
 impl AccountSubscriber {
     pub fn new(
         ws_url: String,
         tracker: Arc<CompressibleAccountTracker>,
-        shutdown_rx: oneshot::Receiver<()>,
+        shutdown_rx: broadcast::Receiver<()>,
     ) -> Self {
         Self {
             ws_url,
@@ -45,7 +54,6 @@ impl AccountSubscriber {
             .map_err(|e| anyhow::anyhow!("Failed to connect to WebSocket: {}", e))?;
 
         let program_id = Pubkey::new_from_array(CTOKEN_PROGRAM_ID);
-
         // Subscribe to compressed token program accounts with filter for compressible account size
         let (mut subscription, unsubscribe) = pubsub_client
             .program_subscribe(
@@ -87,7 +95,7 @@ impl AccountSubscriber {
                         }
                     }
                 }
-                _ = &mut self.shutdown_rx => {
+                _ = self.shutdown_rx.recv() => {
                     info!("Shutdown signal received");
                     unsubscribe().await;
                     break;
@@ -154,3 +162,125 @@ impl AccountSubscriber {
         }
     }
 }
+
+/// Subscribes to registry program logs to detect compress_and_close operations
+/// and remove closed accounts from the tracker by parsing log messages directly
+pub struct LogSubscriber {
+    ws_url: String,
+    tracker: Arc<CompressibleAccountTracker>,
+    shutdown_rx: broadcast::Receiver<()>,
+}
+
+impl LogSubscriber {
+    pub fn new(
+        ws_url: String,
+        tracker: Arc<CompressibleAccountTracker>,
+        shutdown_rx: broadcast::Receiver<()>,
+    ) -> Self {
+        Self {
+            ws_url,
+            tracker,
+            shutdown_rx,
+        }
+    }
+
+    pub async fn run(&mut self) -> Result<()> {
+        info!("Starting log subscriber at {}", self.ws_url);
+
+        // Connect to WebSocket
+        let pubsub_client = PubsubClient::new(&self.ws_url)
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to connect to WebSocket: {}", e))?;
+
+        let registry_program_id = Pubkey::from_str(REGISTRY_PROGRAM_ID_STR)
+            .map_err(|e| anyhow::anyhow!("Invalid registry program ID: {}", e))?;
+
+        // Subscribe to logs mentioning the registry program
+        let filter = RpcTransactionLogsFilter::Mentions(vec![registry_program_id.to_string()]);
+        let config = RpcTransactionLogsConfig {
+            commitment: Some(CommitmentConfig::confirmed()),
+        };
+
+        let (mut subscription, unsubscribe) = pubsub_client
+            .logs_subscribe(filter, config)
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to subscribe to logs: {}", e))?;
+
+        info!(
+            "Log subscription established for registry program {}",
+            registry_program_id
+        );
+
+        // Process subscription messages
+        loop {
+            tokio::select! {
+                result = subscription.next() => {
+                    match result {
+                        Some(response) => {
+                            self.handle_log_notification(response);
+                        }
+                        None => {
+                            error!("Log subscription stream closed unexpectedly");
+                            unsubscribe().await;
+                            return Err(anyhow::anyhow!("Log subscription stream closed"));
+                        }
+                    }
+                }
+                _ = self.shutdown_rx.recv() => {
+                    info!("Shutdown signal received for log subscriber");
+                    unsubscribe().await;
+                    break;
+                }
+            }
+        }
+
+        info!("Log subscriber stopped");
+        Ok(())
+    }
+
+    fn handle_log_notification(&self, response: RpcResponse<RpcLogsResponse>) {
+        let logs_response = response.value;
+
+        // Skip failed transactions
+        if logs_response.err.is_some() {
+            debug!("Skipping failed transaction {}", logs_response.signature);
+            return;
+        }
+
+        // Parse logs looking for compress_and_close entries
+        let mut removed_count = 0;
+        for log in &logs_response.logs {
+            // Look for our log prefix: "Program log: compress_and_close:<pubkey>"
+            // The actual log format is "Program log: compress_and_close:<pubkey>"
+            if let Some(pubkey_str) = log
+                .strip_prefix("Program log: ")
+                .and_then(|s| s.strip_prefix(COMPRESS_AND_CLOSE_LOG_PREFIX))
+            {
+                match Pubkey::from_str(pubkey_str) {
+                    Ok(pubkey) => {
+                        if self.tracker.remove(&pubkey).is_some() {
+                            debug!(
+                                "Removed closed account {} from tracker (compress_and_close log)",
+                                pubkey
+                            );
+                            removed_count += 1;
+                        }
+                    }
+                    Err(e) => {
+                        error!(
+                            "Invalid pubkey in compress_and_close log '{}': {}",
+                            pubkey_str, e
+                        );
+                    }
+                }
+            }
+        }
+
+        if removed_count > 0 {
+            info!(
+                "Removed {} closed accounts from transaction {}",
+                removed_count, logs_response.signature
+            );
+        }
+    }
+}