feat: isolate and track forester worker concurrency

Author: sergeytimoshin

forester/src/epoch_manager.rs

@@ -440,21 +440,19 @@ pub async fn metrics_handler() -> Result<impl warp::Reply> {
     if let Err(e) = encoder.encode(&REGISTRY.gather(), &mut buffer) {
         error!("could not encode custom metrics: {}", e);
     };
-    let mut res = String::from_utf8(buffer.clone()).unwrap_or_else(|e| {
+    let mut res = String::from_utf8(buffer).unwrap_or_else(|e| {
         error!("custom metrics could not be from_utf8'd: {}", e);
         String::new()
     });
-    buffer.clear();
 
     let mut buffer = Vec::new();
     if let Err(e) = encoder.encode(&prometheus::gather(), &mut buffer) {
         error!("could not encode prometheus metrics: {}", e);
     };
-    let res_prometheus = String::from_utf8(buffer.clone()).unwrap_or_else(|e| {
+    let res_prometheus = String::from_utf8(buffer).unwrap_or_else(|e| {
         error!("prometheus metrics could not be from_utf8'd: {}", e);
         String::new()
     });
-    buffer.clear();
 
     res.push_str(&res_prometheus);
     Ok(res)

forester/src/metrics.rs

@@ -208,7 +208,7 @@ pub async fn request_priority_fee_estimate(
         .map_err(|error| PriorityFeeEstimateError::ClientBuild(error.clone()))?;
 
     let response = http_client
-        .post(url.clone())
+        .post(url.as_str())
         .header("Content-Type", "application/json")
         .json(&rpc_request)
         .send()

forester/src/priority_fee.rs

@@ -11,7 +11,10 @@ use light_client::{
 };
 use light_hasher::hash_chain::create_hash_chain_from_slice;
 
-use crate::processor::v2::{common::clamp_to_u16, BatchContext};
+use crate::{
+    logging::should_emit_rate_limited_warning,
+    processor::v2::{common::clamp_to_u16, BatchContext},
+};
 
 pub(crate) fn lock_recover<'a, T>(mutex: &'a Mutex<T>, name: &'static str) -> MutexGuard<'a, T> {
     match mutex.lock() {
@@ -496,71 +499,57 @@ impl StreamingAddressQueue {
         if available < end || start >= end {
             return Ok(None);
         }
-        let actual_end = end;
         let data = lock_recover(&self.data, "streaming_address_queue.data");
 
-        let min_len = [
-            data.addresses.len(),
-            data.low_element_values.len(),
-            data.low_element_next_values.len(),
-            data.low_element_indices.len(),
-            data.low_element_next_indices.len(),
-        ]
-        .into_iter()
-        .min()
-        .unwrap_or(0);
-        if min_len < actual_end {
-            return Err(anyhow!(
-                "incomplete batch data: min field length {} < required end {}",
-                min_len,
-                actual_end
-            ));
-        }
-
-        let addresses = data.addresses[start..actual_end].to_vec();
-        if addresses.is_empty() {
-            return Ok(None);
-        }
-        let expected_len = addresses.len();
-        let Some(low_element_values) = data
-            .low_element_values
-            .get(start..end)
-            .map(|slice| slice.to_vec())
-        else {
-            return Ok(None);
-        };
-        let Some(low_element_next_values) = data
-            .low_element_next_values
-            .get(start..end)
-            .map(|slice| slice.to_vec())
-        else {
-            return Ok(None);
-        };
-        let Some(low_element_indices) = data
-            .low_element_indices
-            .get(start..end)
-            .map(|slice| slice.to_vec())
+        // `available` can be bumped before every parallel array is filled,
+        // so a missing range here means "not ready yet" — return Ok(None)
+        // and let the caller retry on the next tick.
+        let range = start..end;
+        let (
+            Some(addresses),
+            Some(low_element_values),
+            Some(low_element_next_values),
+            Some(low_element_indices),
+            Some(low_element_next_indices),
+        ) = (
+            data.addresses.get(range.clone()).map(<[_]>::to_vec),
+            data.low_element_values
+                .get(range.clone())
+                .map(<[_]>::to_vec),
+            data.low_element_next_values
+                .get(range.clone())
+                .map(<[_]>::to_vec),
+            data.low_element_indices
+                .get(range.clone())
+                .map(<[_]>::to_vec),
+            data.low_element_next_indices
+                .get(range.clone())
+                .map(<[_]>::to_vec),
+        )
         else {
             return Ok(None);
         };
-        let Some(low_element_next_indices) = data
-            .low_element_next_indices
-            .get(start..end)
-            .map(|slice| slice.to_vec())
-        else {
-            return Ok(None);
+
+        // Proofs can also be unavailable if the indexer hasn't populated the
+        // merkle nodes for this range yet — return Ok(None) and retry. Log
+        // at warn (rate-limited) so persistent failures are still visible.
+        let low_element_proofs = match data.reconstruct_proofs::<HEIGHT>(range) {
+            Ok(proofs) => proofs,
+            Err(error) => {
+                if should_emit_rate_limited_warning(
+                    "address_queue_proofs_not_ready",
+                    std::time::Duration::from_secs(60),
+                ) {
+                    tracing::warn!(
+                        ?error,
+                        start,
+                        end,
+                        "address proof reconstruction not ready, retrying"
+                    );
+                }
+                return Ok(None);
+            }
         };
-        if [
-            low_element_values.len(),
-            low_element_next_values.len(),
-            low_element_indices.len(),
-            low_element_next_indices.len(),
-        ]
-        .iter()
-        .any(|&len| len != expected_len)
-        {
-            return Ok(None);
-        }
 
         let leaves_hashchain = match data.leaves_hash_chains.get(hashchain_idx).copied() {
             Some(hashchain) => hashchain,
@@ -582,15 +571,11 @@ impl StreamingAddressQueue {
         };
 
         Ok(Some(AddressBatchSnapshot {
-            low_element_values: data.low_element_values[start..actual_end].to_vec(),
-            low_element_next_values: data.low_element_next_values[start..actual_end].to_vec(),
-            low_element_indices: data.low_element_indices[start..actual_end].to_vec(),
-            low_element_next_indices: data.low_element_next_indices[start..actual_end].to_vec(),
-            low_element_proofs: data
-                .reconstruct_proofs::<HEIGHT>(start..actual_end)
-                .map_err(|error| {
-                    anyhow!("incomplete batch data: failed to reconstruct proofs: {error}")
-                })?,
+            low_element_values,
+            low_element_next_values,
+            low_element_indices,
+            low_element_next_indices,
+            low_element_proofs,
             addresses,
             leaves_hashchain,
         }))

forester/src/processor/v2/helpers.rs

@@ -139,19 +139,19 @@ impl ProofClients {
                 config.polling_interval,
                 config.max_wait_time,
                 config.api_key.clone(),
-            )?,
+            ),
             nullify_client: ProofClient::with_config(
                 config.update_url.clone(),
                 config.polling_interval,
                 config.max_wait_time,
                 config.api_key.clone(),
-            )?,
+            ),
             address_append_client: ProofClient::with_config(
                 config.address_append_url.clone(),
                 config.polling_interval,
                 config.max_wait_time,
                 config.api_key.clone(),
-            )?,
+            ),
         })
     }
 

forester/src/processor/v2/proof_worker.rs

@@ -23,9 +23,7 @@ use light_compressed_account::{
 use light_compressed_token::process_transfer::{
     transfer_sdk::create_transfer_instruction, TokenTransferOutputData,
 };
-use light_token::compat::TokenDataWithMerkleContext;
 use light_program_test::accounts::test_accounts::TestAccounts;
-use light_prover_client::prover::spawn_prover;
 use light_registry::{
     protocol_config::state::{ProtocolConfig, ProtocolConfigPda},
     utils::get_protocol_config_pda_address,
@@ -34,6 +32,7 @@ use light_test_utils::{
     conversions::sdk_to_program_token_data, spl::create_mint_helper_with_keypair,
     system_program::create_invoke_instruction,
 };
+use light_token::compat::TokenDataWithMerkleContext;
 use rand::{prelude::SliceRandom, rngs::StdRng, Rng, SeedableRng};
 use serial_test::serial;
 use solana_program::{native_token::LAMPORTS_PER_SOL, pubkey::Pubkey};
@@ -87,7 +86,6 @@ async fn test_state_indexer_async_batched() {
         validator_args: vec![],
     }))
     .await;
-    spawn_prover().await;
 
     let env = TestAccounts::get_local_test_validator_accounts();
     let mut config = forester_config();
@@ -306,10 +304,7 @@ async fn wait_for_slot(rpc: &mut LightClient, target_slot: u64) {
             return;
         }
         Err(e) => {
-            println!(
-                "warp_to_slot unavailable ({}), falling back to polling",
-                e
-            );
+            println!("warp_to_slot unavailable ({}), falling back to polling", e);
         }
     }
     while rpc.get_slot().await.unwrap() < target_slot {

forester/tests/legacy/batched_state_async_indexer_test.rs

@@ -39,7 +39,6 @@ pub enum ProverClientError {
 
     #[error("Integer conversion failed: {0}")]
     IntegerConversion(String),
-
     #[error("Hashchain mismatch: computed {computed:?} != expected {expected:?} (batch_size={batch_size}, next_index={next_index})")]
     HashchainMismatch {
         computed: [u8; 32],

prover/client/src/errors.rs

@@ -2,7 +2,7 @@ use std::process::Command;
 
 use light_hasher::{Hasher, Poseidon};
 use light_sparse_merkle_tree::changelog::ChangelogEntry;
-use num_bigint::{BigInt, BigUint};
+use num_bigint::{BigInt, BigUint, Sign};
 use num_traits::{Num, ToPrimitive};
 use serde::Serialize;
 
@@ -35,10 +35,17 @@ pub fn convert_endianness_128(bytes: &[u8]) -> Vec<u8> {
         .collect::<Vec<u8>>()
 }
 
-pub fn bigint_to_u8_32(n: &BigInt) -> Result<[u8; 32], Box<dyn std::error::Error>> {
-    let (_, bytes_be) = n.to_bytes_be();
+pub fn bigint_to_u8_32(n: &BigInt) -> Result<[u8; 32], ProverClientError> {
+    let (sign, bytes_be) = n.to_bytes_be();
+    if sign == Sign::Minus {
+        return Err(ProverClientError::InvalidProofData(
+            "negative integers are not valid field elements".to_string(),
+        ));
+    }
     if bytes_be.len() > 32 {
-        Err("Number too large to fit in [u8; 32]")?;
+        return Err(ProverClientError::InvalidProofData(
+            "number too large to fit in [u8; 32]".to_string(),
+        ));
     }
     let mut array = [0; 32];
     let bytes = &bytes_be[..bytes_be.len()];

prover/client/src/helpers.rs

@@ -84,21 +84,21 @@ impl ProofClient {
         polling_interval: Duration,
         max_wait_time: Duration,
         api_key: Option<String>,
-    ) -> Result<Self, ProverClientError> {
+    ) -> Self {
         let initial_poll_delay = if api_key.is_some() {
             Duration::from_millis(INITIAL_POLL_DELAY_LARGE_CIRCUIT_MS)
         } else {
             Duration::from_millis(INITIAL_POLL_DELAY_SMALL_CIRCUIT_MS)
         };
 
-        Ok(Self {
+        Self {
             client: build_http_client(),
             server_address,
             polling_interval,
             max_wait_time,
             api_key,
             initial_poll_delay,
-        })
+        }
     }
 
     #[allow(unused)]