refactor: removed mutexed indexer from forester, instead use rpc.indexer() (#1864)

* refactor: removed mutexed indexer from forester, instead use rpc.indexer()

* refactor: simplify generic constraints in instruction stream functions

* Remove unused imports in tests

* ignore legacy e2e test (photon needs to support custom trees before we
can re-enable it)

* Potential fix for code scanning alert no. 102: Workflow does not contain permissions

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

---------

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

Author: sergeytimoshin

.github/workflows/forester-tests.yml

@@ -1,5 +1,8 @@
 name: forester-tests
 
+permissions:
+  contents: read
+
 on:
   push:
     branches: [main]
@@ -38,26 +41,9 @@ env:
 
 jobs:
   test:
-    strategy:
-      matrix:
-        test-name:
-          [
-            {
-              name: "e2e (legacy)",
-              command: "test_e2e_v1",
-              timeout: 60,
-              needs-test-program: false,
-            },
-            {
-              name: "e2e",
-              command: "test_e2e_v2",
-              timeout: 60,
-              needs-test-program: true,
-            },
-          ]
-    name: test-${{ matrix.test-name.name }}
+    name: Forester e2e test
     runs-on: warp-ubuntu-latest-x64-4x
-    timeout-minutes: ${{ matrix.test-name.timeout }}
+    timeout-minutes: 30
 
     services:
       redis:
@@ -75,31 +61,35 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
+
       - name: Setup and build
         uses: ./.github/actions/setup-and-build
         with:
           skip-components: "redis"
+
       - name: Clean build artifacts before tests
         shell: bash
         run: |
           cargo clean
           rm -rf target/debug/deps/*
+
       - name: Check available disk space
         shell: bash
         run: |
           df -h /
           du -sh /home/runner/work/* | sort -hr | head -n 10
+
       - name: Build CLI
         run: |
           source ./scripts/devenv.sh
           npx nx build @lightprotocol/zk-compression-cli
+
       - name: Build test program
-        if: ${{ matrix.test-name.needs-test-program }}
         run: |
           source ./scripts/devenv.sh
           cargo test-sbf -p create-address-test-program
 
-      - name: Run ${{ matrix.test-name.name }} tests
+      - name: Test
         run: |
           source ./scripts/devenv.sh
-          cargo test --package forester ${{ matrix.test-name.command }} -- --nocapture
+          cargo test --package forester e2e_test -- --nocapture

forester-utils/src/error.rs

@@ -1,4 +1,5 @@
 use light_batched_merkle_tree::errors::BatchedMerkleTreeError;
+use light_client::rpc::RpcError;
 use light_hasher::HasherError;
 use thiserror::Error;
 
@@ -10,8 +11,8 @@ pub enum ForesterUtilsError {
     Parse(String),
     #[error("prover error: {0:?}")]
     Prover(String),
-    #[error("rpc error: {0:?}")]
-    Rpc(String),
+    #[error("rpc error")]
+    Rpc(#[from] RpcError),
     #[error("indexer error: {0:?}")]
     Indexer(String),
     #[error("invalid slot number")]
@@ -22,9 +23,6 @@ pub enum ForesterUtilsError {
     #[error("Account zero-copy error: {0}")]
     AccountZeroCopy(String),
 
-    #[error("light client error: {0}")]
-    LightClient(#[from] light_client::rpc::RpcError),
-
     #[error("batched merkle tree error: {0}")]
     BatchedMerkleTree(#[from] BatchedMerkleTreeError),
 

forester-utils/src/instructions/address_batch_update.rs

@@ -19,20 +19,14 @@ use light_prover_client::{
     proof_types::batch_address_append::get_batch_address_append_circuit_inputs,
 };
 use light_sparse_merkle_tree::SparseMerkleTree;
-use tokio::sync::Mutex;
 use tracing::{debug, error, info, warn};
 
-use crate::{error::ForesterUtilsError, rpc_pool::SolanaRpcPool, utils::wait_for_indexer};
+use crate::{error::ForesterUtilsError, rpc_pool::SolanaRpcPool};
 
 const MAX_PHOTON_ELEMENTS_PER_CALL: usize = 500;
 
-pub struct AddressUpdateConfig<R, I>
-where
-    R: Rpc + Send + Sync,
-    I: Indexer + Send,
-{
+pub struct AddressUpdateConfig<R: Rpc> {
     pub rpc_pool: Arc<SolanaRpcPool<R>>,
-    pub indexer: Arc<Mutex<I>>,
     pub merkle_tree_pubkey: Pubkey,
     pub prover_url: String,
     pub polling_interval: Duration,
@@ -41,8 +35,8 @@ where
 }
 
 #[allow(clippy::too_many_arguments)]
-fn stream_instruction_data<'a, I>(
-    indexer: Arc<Mutex<I>>,
+async fn stream_instruction_data<'a, R: Rpc>(
+    rpc_pool: Arc<SolanaRpcPool<R>>,
     merkle_tree_pubkey: Pubkey,
     prover_url: String,
     polling_interval: Duration,
@@ -53,24 +47,24 @@ fn stream_instruction_data<'a, I>(
     mut current_root: [u8; 32],
     yield_batch_size: usize,
 ) -> impl Stream<Item = Result<Vec<InstructionDataAddressAppendInputs>, ForesterUtilsError>> + Send + 'a
-where
-    I: Indexer + Send + 'a,
 {
     stream! {
         let proof_client = Arc::new(ProofClient::with_config(prover_url, polling_interval, max_wait_time));
         let max_zkp_batches_per_call = calculate_max_zkp_batches_per_call(zkp_batch_size);
         let total_chunks = leaves_hash_chains.len().div_ceil(max_zkp_batches_per_call);
 
         for chunk_idx in 0..total_chunks {
-            let mut indexer_guard = indexer.lock().await;
             let chunk_start = chunk_idx * max_zkp_batches_per_call;
             let chunk_end = std::cmp::min(chunk_start + max_zkp_batches_per_call, leaves_hash_chains.len());
             let chunk_hash_chains = &leaves_hash_chains[chunk_start..chunk_end];
 
             let elements_for_chunk = chunk_hash_chains.len() * zkp_batch_size as usize;
             let processed_items_offset = chunk_start * zkp_batch_size as usize;
 
-            let indexer_update_info = match indexer_guard
+            let indexer_update_info = {
+                let mut connection = rpc_pool.get_connection().await?;
+                let indexer = connection.indexer_mut()?;
+                match indexer
                 .get_address_queue_with_proofs(
                     &merkle_tree_pubkey,
                     elements_for_chunk as u16,
@@ -83,7 +77,8 @@ where
                         yield Err(ForesterUtilsError::Indexer(format!("Failed to get address queue with proofs: {}", e)));
                         return;
                     }
-                };
+                }
+            };
 
             if chunk_idx == 0 {
                 if let Some(first_proof) = indexer_update_info.value.non_inclusion_proofs.first() {
@@ -296,8 +291,8 @@ fn get_all_circuit_inputs_for_chunk(
     Ok((all_inputs, current_root))
 }
 
-pub async fn get_address_update_instruction_stream<'a, R, I>(
-    config: AddressUpdateConfig<R, I>,
+pub async fn get_address_update_instruction_stream<'a, R: Rpc>(
+    config: AddressUpdateConfig<R>,
     merkle_tree_data: crate::ParsedMerkleTreeData,
 ) -> Result<
     (
@@ -312,17 +307,7 @@ pub async fn get_address_update_instruction_stream<'a, R, I>(
         u16,
     ),
     ForesterUtilsError,
->
-where
-    R: Rpc + Send + Sync + 'a,
-    I: Indexer + Send + 'a,
-{
-    let rpc = config.rpc_pool.get_connection().await?;
-    let indexer_guard = config.indexer.lock().await;
-    wait_for_indexer(&*rpc, &*indexer_guard).await?;
-    drop(rpc);
-    drop(indexer_guard);
-
+> {
     let (current_root, leaves_hash_chains, start_index, zkp_batch_size) = (
         merkle_tree_data.current_root,
         merkle_tree_data.leaves_hash_chains,
@@ -336,7 +321,7 @@ where
     }
 
     let stream = stream_instruction_data(
-        config.indexer,
+        config.rpc_pool,
         config.merkle_tree_pubkey,
         config.prover_url,
         config.polling_interval,
@@ -346,7 +331,8 @@ where
         zkp_batch_size,
         current_root,
         config.ixs_per_tx,
-    );
+    )
+    .await;
 
     Ok((Box::pin(stream), zkp_batch_size))
 }

forester-utils/src/instructions/state_batch_append.rs

@@ -18,7 +18,6 @@ use light_prover_client::{
     proof_types::batch_append::{get_batch_append_inputs, BatchAppendsCircuitInputs},
 };
 use light_sparse_merkle_tree::changelog::ChangelogEntry;
-use tokio::sync::Mutex;
 use tracing::trace;
 
 use crate::{
@@ -45,9 +44,8 @@ async fn generate_zkp_proof(
 }
 
 #[allow(clippy::too_many_arguments)]
-pub async fn get_append_instruction_stream<'a, R, I>(
+pub async fn get_append_instruction_stream<'a, R: Rpc>(
     rpc_pool: Arc<SolanaRpcPool<R>>,
-    indexer: Arc<Mutex<I>>,
     merkle_tree_pubkey: Pubkey,
     prover_url: String,
     polling_interval: Duration,
@@ -67,16 +65,8 @@ pub async fn get_append_instruction_stream<'a, R, I>(
         u16,
     ),
     ForesterUtilsError,
->
-where
-    R: Rpc + Send + Sync + 'a,
-    I: Indexer + Send + 'a,
-{
+> {
     trace!("Initializing append batch instruction stream with parsed data");
-
-    let (indexer_guard, rpc_result) = tokio::join!(indexer.lock(), rpc_pool.get_connection());
-    let rpc = rpc_result?;
-
     let (merkle_tree_next_index, mut current_root, _) = (
         merkle_tree_data.next_index,
         merkle_tree_data.current_root,
@@ -91,19 +81,18 @@ where
         trace!("No hash chains to process, returning empty stream.");
         return Ok((Box::pin(futures::stream::empty()), zkp_batch_size));
     }
-
-    wait_for_indexer(&*rpc, &*indexer_guard).await?;
+    let rpc = rpc_pool.get_connection().await?;
+    wait_for_indexer(&*rpc).await?;
     drop(rpc);
-    drop(indexer_guard);
 
     let stream = stream! {
         let total_elements = zkp_batch_size as usize * leaves_hash_chains.len();
         let offset = merkle_tree_next_index;
 
         let queue_elements = {
-            let mut indexer_guard = indexer.lock().await;
-
-            match indexer_guard
+            let mut connection = rpc_pool.get_connection().await?;
+            let indexer = connection.indexer_mut()?;
+            match indexer
                 .get_queue_elements(
                     merkle_tree_pubkey.to_bytes(),
                     QueueType::OutputStateV2,

forester-utils/src/instructions/state_batch_nullify.rs

@@ -17,7 +17,6 @@ use light_prover_client::{
     proof_client::ProofClient,
     proof_types::batch_update::{get_batch_update_inputs, BatchUpdateCircuitInputs},
 };
-use tokio::sync::Mutex;
 use tracing::{debug, trace};
 
 use crate::{
@@ -44,9 +43,8 @@ async fn generate_nullify_zkp_proof(
 }
 
 #[allow(clippy::too_many_arguments)]
-pub async fn get_nullify_instruction_stream<'a, R, I>(
+pub async fn get_nullify_instruction_stream<'a, R: Rpc>(
     rpc_pool: Arc<SolanaRpcPool<R>>,
-    indexer: Arc<Mutex<I>>,
     merkle_tree_pubkey: Pubkey,
     prover_url: String,
     polling_interval: Duration,
@@ -66,13 +64,7 @@ pub async fn get_nullify_instruction_stream<'a, R, I>(
         u16,
     ),
     ForesterUtilsError,
->
-where
-    R: Rpc + Send + Sync + 'a,
-    I: Indexer + Send + 'a,
-{
-    let rpc = rpc_pool.get_connection().await?;
-
+> {
     let (mut current_root, leaves_hash_chains, num_inserted_zkps, zkp_batch_size) = (
         merkle_tree_data.current_root,
         merkle_tree_data.leaves_hash_chains,
@@ -85,10 +77,9 @@ where
         return Ok((Box::pin(futures::stream::empty()), zkp_batch_size));
     }
 
-    let indexer_guard = indexer.lock().await;
-    wait_for_indexer(&*rpc, &*indexer_guard).await?;
+    let rpc = rpc_pool.get_connection().await?;
+    wait_for_indexer(&*rpc).await?;
     drop(rpc);
-    drop(indexer_guard);
 
     let stream = stream! {
         let total_elements = zkp_batch_size as usize * leaves_hash_chains.len();
@@ -97,9 +88,9 @@ where
         trace!("Requesting {} total elements with offset {}", total_elements, offset);
 
         let all_queue_elements = {
-            let mut indexer_guard = indexer.lock().await;
-            indexer_guard
-            .get_queue_elements(
+            let mut connection = rpc_pool.get_connection().await?;
+            let indexer = connection.indexer_mut()?;
+            indexer.get_queue_elements(
                 merkle_tree_pubkey.to_bytes(),
                 QueueType::InputStateV2,
                 total_elements as u16,

forester-utils/src/utils.rs

@@ -28,16 +28,10 @@ pub async fn airdrop_lamports<R: Rpc>(
     Ok(())
 }
 
-pub async fn wait_for_indexer<R: Rpc, I: Indexer>(
-    rpc: &R,
-    indexer: &I,
-) -> Result<(), ForesterUtilsError> {
-    let rpc_slot = rpc
-        .get_slot()
-        .await
-        .map_err(|_| ForesterUtilsError::Rpc("Failed to get rpc slot".into()))?;
+pub async fn wait_for_indexer<R: Rpc>(rpc: &R) -> Result<(), ForesterUtilsError> {
+    let rpc_slot = rpc.get_slot().await?;
 
-    let indexer_slot = indexer.get_indexer_slot(None).await;
+    let indexer_slot = rpc.indexer()?.get_indexer_slot(None).await;
 
     let mut indexer_slot = match indexer_slot {
         Ok(slot) => slot,
@@ -66,7 +60,7 @@ pub async fn wait_for_indexer<R: Rpc, I: Indexer>(
 
         tokio::task::yield_now().await;
         sleep(std::time::Duration::from_millis(500)).await;
-        indexer_slot = indexer.get_indexer_slot(None).await.map_err(|e| {
+        indexer_slot = rpc.indexer()?.get_indexer_slot(None).await.map_err(|e| {
             error!("failed to get indexer slot from indexer: {:?}", e);
             ForesterUtilsError::Indexer("Failed to get indexer slot".into())
         })?;

forester/package.json

@@ -4,12 +4,7 @@
   "license": "GPL-3.0",
   "scripts": {
     "build": "cargo build",
-    "test": "RUSTFLAGS=\"--cfg tokio_unstable -D warnings\" cargo test --package forester -- --nocapture",
-    "e2e-test": "source .env && RUST_LOG=forester=debug,forester_utils=debug cargo test --package forester test_e2e_v2 -- --nocapture",
-    "test-address-batched-local": "RUST_LOG=forester=debug,forester_utils=debug cargo test --package forester test_address_batched -- --nocapture",
-    "e2e-v1": "RUST_LOG=forester=debug,forester_utils=debug cargo test --package forester test_e2e_v1 -- --nocapture",
-    "test-address-v2": "RUST_LOG=forester=debug,forester_utils=debug cargo test --package forester test_create_v2_address -- --nocapture",
-    "test-state-batched-indexer-async": "RUST_LOG=forester=debug,forester_utils=debug cargo test --package forester test_state_indexer_async_batched -- --nocapture",
+    "test": "source .env && RUST_LOG=forester=debug,forester_utils=debug cargo test --package forester test_e2e_v2 -- --nocapture",
     "docker:build": "docker build --tag forester -f Dockerfile .."
   },
   "devDependencies": {