feat: forester: do not retry on not eligible slots (#1089)

* Introduced a new error variant `NotEligible` in `ForesterError`. Updated the eligibility check to return this error and handle it appropriately during the transaction processing loop.

* format

* Refactor retry delay calculation to prevent overflows

Modified the retry delay calculation by using `saturating_mul` and `saturating_pow` to avoid potential overflow issues. This ensures the application remains stable even if the retry count becomes excessively high.

* format

Author: sergeytimoshin

forester/src/epoch_manager.rs

@@ -720,7 +720,7 @@ impl<R: RpcConnection, I: Indexer<R>> EpochManager<R, I> {
         &self,
         registration_info: &ForesterEpochInfo,
         tree_account: &TreeAccounts,
-    ) -> Result<bool> {
+    ) -> Result<()> {
         let mut rpc = self.rpc_pool.get_connection().await?;
         let current_slot = rpc.get_slot().await?;
         let forester_epoch_pda = rpc
@@ -754,7 +754,11 @@ impl<R: RpcConnection, I: Indexer<R>> EpochManager<R, I> {
             "tree_schedule.slots[{}] = {:?}",
             light_slot, tree_schedule.slots[light_slot as usize]
         );
-        Ok(tree_schedule.is_eligible(light_slot))
+        if tree_schedule.is_eligible(light_slot) {
+            Ok(())
+        } else {
+            Err(ForesterError::NotEligible)
+        }
     }
 
     async fn process_transaction_batch_with_retry(
@@ -764,60 +768,68 @@ impl<R: RpcConnection, I: Indexer<R>> EpochManager<R, I> {
         proof_chunk: &[Proof],
         indexer_chunk: &[WorkItem],
     ) -> Result<Option<Signature>> {
-        let first_work_item = indexer_chunk
+        let work_item = indexer_chunk
             .first()
             .ok_or_else(|| ForesterError::Custom("Empty indexer chunk".to_string()))?;
         debug!(
             "Processing work item {:?} with {} instructions",
-            first_work_item.queue_item_data.hash,
+            work_item.queue_item_data.hash,
             transaction_chunk.len()
         );
         const BASE_RETRY_DELAY: Duration = Duration::from_millis(100);
 
         let mut retries = 0;
         loop {
-            if !self
-                .check_eligibility(epoch_info, &first_work_item.tree_account)
-                .await?
-            {
-                debug!("Forester not eligible for this slot, skipping batch");
-                return Ok(None);
-            }
-
             match self
-                .process_transaction_batch(
-                    epoch_info,
-                    transaction_chunk,
-                    proof_chunk,
-                    indexer_chunk,
-                )
+                .check_eligibility(epoch_info, &work_item.tree_account)
                 .await
             {
-                Ok(signature) => {
-                    debug!(
-                        "Work item {:?} processed successfully. Signature: {:?}",
-                        first_work_item.queue_item_data.hash, signature
-                    );
-                    self.increment_processed_items_count(epoch_info.epoch.epoch)
-                        .await;
-                    return Ok(Some(signature));
+                Ok(_) => {
+                    match self
+                        .process_transaction_batch(
+                            epoch_info,
+                            transaction_chunk,
+                            proof_chunk,
+                            indexer_chunk,
+                        )
+                        .await
+                    {
+                        Ok(signature) => {
+                            debug!(
+                                "Work item {:?} processed successfully. Signature: {:?}",
+                                work_item.queue_item_data.hash, signature
+                            );
+                            self.increment_processed_items_count(epoch_info.epoch.epoch)
+                                .await;
+                            return Ok(Some(signature));
+                        }
+                        Err(e) => {
+                            if retries >= self.config.max_retries {
+                                error!(
+                                    "Max retries reached for work item {:?}. Error: {:?}",
+                                    work_item.queue_item_data.hash, e
+                                );
+                                return Err(e);
+                            }
+                            let delay = BASE_RETRY_DELAY
+                                .saturating_mul(2u32.saturating_pow(retries as u32));
+                            let jitter = rand::thread_rng().gen_range(0..=50);
+                            sleep(delay + Duration::from_millis(jitter)).await;
+                            retries += 1;
+                            warn!(
+                                "Retrying work item {:?}. Attempt {}/{}",
+                                work_item.queue_item_data.hash, retries, self.config.max_retries
+                            );
+                        }
+                    }
+                }
+                Err(ForesterError::NotEligible) => {
+                    debug!("Forester not eligible for this slot, skipping batch");
+                    return Ok(None);
                 }
                 Err(e) => {
-                    if retries >= self.config.max_retries {
-                        error!(
-                            "Max retries reached for work item {:?}. Error: {:?}",
-                            first_work_item.queue_item_data.hash, e
-                        );
-                        return Err(e);
-                    }
-                    let delay = BASE_RETRY_DELAY * 2u32.pow(retries as u32);
-                    let jitter = rand::thread_rng().gen_range(0..=50);
-                    sleep(delay + Duration::from_millis(jitter)).await;
-                    retries += 1;
-                    warn!(
-                        "Retrying work item {:?}. Attempt {}/{}",
-                        first_work_item.queue_item_data.hash, retries, self.config.max_retries
-                    );
+                    error!("Error checking eligibility: {:?}", e);
+                    return Err(e);
                 }
             }
         }

forester/src/errors.rs

@@ -12,6 +12,8 @@ use tokio::task::JoinError;
 
 #[derive(Error, Debug)]
 pub enum ForesterError {
+    #[error("Element is not eligible for foresting")]
+    NotEligible,
     #[error("RPC Error: {0}")]
     RpcError(#[from] RpcError),
     #[error("failed to deserialize account data")]
@@ -55,6 +57,7 @@ pub enum ForesterError {
 impl Clone for ForesterError {
     fn clone(&self) -> Self {
         match self {
+            ForesterError::NotEligible => ForesterError::NotEligible,
             ForesterError::RpcError(_) => ForesterError::Custom("RPC Error".to_string()),
             ForesterError::DeserializeError(e) => ForesterError::DeserializeError(e.clone()),
             ForesterError::CopyMerkleTreeError(_) => {
@@ -89,6 +92,7 @@ impl Clone for ForesterError {
 impl ForesterError {
     pub fn to_owned(&self) -> Self {
         match self {
+            ForesterError::NotEligible => ForesterError::NotEligible,
             ForesterError::RpcError(e) => ForesterError::Custom(format!("RPC Error: {:?}", e)),
             ForesterError::DeserializeError(e) => {
                 ForesterError::Custom(format!("Deserialize Error: {:?}", e))