feat: state updates via single pipeline + StagingTree

forester-utils/Cargo.toml

@@ -21,6 +21,7 @@ light-indexed-merkle-tree = { workspace = true }
 light-compressed-account = { workspace = true, features = ["std"] }
 light-batched-merkle-tree = { workspace = true }
 light-merkle-tree-metadata = { workspace = true }
+light-merkle-tree-reference = { workspace = true }
 light-sparse-merkle-tree = { workspace = true }
 light-account-checks = { workspace = true }
 light-sdk = { workspace = true }
@@ -39,6 +40,7 @@ anchor-lang = { workspace = true }
 solana-sdk = { workspace = true }
 
 thiserror = { workspace = true }
+anyhow = { workspace = true }
 
 tracing = { workspace = true }
 
@@ -51,3 +53,5 @@ governor = { workspace = true }
 [dev-dependencies]
 tokio-postgres = "0.7"
 bs58 = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }

forester-utils/src/error.rs

@@ -28,4 +28,7 @@ pub enum ForesterUtilsError {
 
     #[error("pool error: {0}")]
     Pool(#[from] PoolError),
+
+    #[error("error: {0}")]
+    StagingTree(String),
 }

forester-utils/src/forester_epoch.rs

@@ -182,6 +182,23 @@ impl TreeForesterSchedule {
     pub fn is_eligible(&self, forester_slot: u64) -> bool {
         self.slots[forester_slot as usize].is_some()
     }
+
+    /// Returns the end solana slot of the last consecutive eligible slot
+    /// starting from the given light slot index.
+    pub fn get_consecutive_eligibility_end(&self, from_slot_idx: usize) -> Option<u64> {
+        let mut last_eligible_end = None;
+
+        for slot_opt in self.slots.iter().skip(from_slot_idx) {
+            match slot_opt {
+                Some(slot) => {
+                    last_eligible_end = Some(slot.end_solana_slot);
+                }
+                None => break,
+            }
+        }
+
+        last_eligible_end
+    }
 }
 
 #[derive(Debug, Clone, AnchorSerialize, AnchorDeserialize, Default, PartialEq, Eq)]

forester-utils/src/instructions/address_batch_update.rs

@@ -94,7 +94,6 @@ async fn stream_instruction_data<'a, R: Rpc>(
                 }
             };
 
-            // Log Photon response details
             debug!(
                 "Photon response for chunk {}: received {} addresses, batch_start_index={}, first_queue_index={:?}, last_queue_index={:?}",
                 chunk_idx,
@@ -104,7 +103,6 @@ async fn stream_instruction_data<'a, R: Rpc>(
                 indexer_update_info.value.addresses.last().map(|a| a.queue_index)
             );
 
-            // Update next_queue_index for the next chunk based on the last address returned
             if let Some(last_address) = indexer_update_info.value.addresses.last() {
                 next_queue_index = Some(last_address.queue_index + 1);
                 debug!(
@@ -143,9 +141,8 @@ async fn stream_instruction_data<'a, R: Rpc>(
             };
             current_root = new_current_root;
 
-            info!("Generating {} zk proofs for batch_address chunk {} (parallel)", all_inputs.len(), chunk_idx + 1);
+            info!("Generating {} zk proofs for batch_address chunk {}", all_inputs.len(), chunk_idx + 1);
 
-            // Generate ALL proofs in parallel using join_all
             let proof_futures: Vec<_> = all_inputs.into_iter().enumerate().map(|(i, inputs)| {
                 let client = Arc::clone(&proof_client);
                 async move {
@@ -154,10 +151,8 @@ async fn stream_instruction_data<'a, R: Rpc>(
                 }
             }).collect();
 
-            // Wait for all proofs to complete in parallel
             let proof_results = futures::future::join_all(proof_futures).await;
 
-            // Process results and batch them into groups of MAX_PROOFS_PER_TX
             let mut proof_buffer = Vec::new();
             for (idx, result) in proof_results {
                 match result {
@@ -172,7 +167,6 @@ async fn stream_instruction_data<'a, R: Rpc>(
                         };
                         proof_buffer.push(instruction_data);
 
-                        // Yield when we have MAX_PROOFS_PER_TX proofs ready
                         if proof_buffer.len() >= MAX_PROOFS_PER_TX {
                             yield Ok(proof_buffer.clone());
                             proof_buffer.clear();

forester-utils/src/instructions/mod.rs

@@ -1,6 +1,4 @@
 pub mod address_batch_update;
 pub mod create_account;
-pub mod state_batch_append;
-pub mod state_batch_nullify;
 
 pub use create_account::create_account_instruction;