fix: 🐛 BSP volunteering issues (#553)

* fix: 🩹 update return of earliest volunteer tick to show actual value

* fix: 🐛 update BSP volunteer task to handle errors better

* refactor: 🔊 update new log from info to debug to not clutter prod logs

* feat: 🔊 Add logs for successful accepts and rejects of MSP

* test: ✅ stop BSPs so they don't volunteer while doing the tests

* feat: 🔊 Add emoji to logs

* feat: ✨ Use `submit_extrinsic_with_retry` to improve volunteering retry

* fix: 🩹 Add missing static

* fix: 🐛 Unvolunteer when extrinsic failed with retries

* fix: 🐛 use ceil division to calculate BSP volunteering threshold

* test: ✅ minor fixes to multiple tests

* fix: 🐛 avoid threshold overflow

---------

Co-authored-by: Facundo Farall <[email protected]>

Author: TDemeco

client/blockchain-service/src/commands.rs

@@ -10,7 +10,8 @@ use sp_api::ApiError;
 
 use pallet_file_system_runtime_api::{
     IsStorageRequestOpenToVolunteersError, QueryBspConfirmChunksToProveForFileError,
-    QueryFileEarliestVolunteerTickError, QueryMspConfirmChunksToProveForFileError,
+    QueryBspsVolunteeredForFileError, QueryFileEarliestVolunteerTickError,
+    QueryMspConfirmChunksToProveForFileError,
 };
 use pallet_payment_streams_runtime_api::GetUsersWithDebtOverThresholdError;
 use pallet_proofs_dealer_runtime_api::{
@@ -92,6 +93,11 @@ pub enum BlockchainServiceCommand<Runtime: StorageEnableRuntime> {
         msp_id: ProofsDealerProviderId<Runtime>,
         file_key: Runtime::Hash,
     },
+    #[command(success_type = bool, error_type = QueryBspsVolunteeredForFileError)]
+    QueryBspVolunteeredForFile {
+        bsp_id: BackupStorageProviderId<Runtime>,
+        file_key: Runtime::Hash,
+    },
     #[command(success_type = Vec<Multiaddr>, error_type = QueryProviderMultiaddressesError)]
     QueryProviderMultiaddresses { provider_id: ProviderId<Runtime> },
     QueueSubmitProofRequest {

client/blockchain-service/src/handler.rs

@@ -17,7 +17,8 @@ use sp_runtime::{traits::Header, SaturatedConversion, Saturating};
 
 use pallet_file_system_runtime_api::{
     FileSystemApi, IsStorageRequestOpenToVolunteersError, QueryBspConfirmChunksToProveForFileError,
-    QueryFileEarliestVolunteerTickError, QueryMspConfirmChunksToProveForFileError,
+    QueryBspsVolunteeredForFileError, QueryFileEarliestVolunteerTickError,
+    QueryMspConfirmChunksToProveForFileError,
 };
 use pallet_payment_streams_runtime_api::{GetUsersWithDebtOverThresholdError, PaymentStreamsApi};
 use pallet_proofs_dealer_runtime_api::{
@@ -617,6 +618,30 @@ where
                         }
                     }
                 }
+                BlockchainServiceCommand::QueryBspVolunteeredForFile {
+                    bsp_id,
+                    file_key,
+                    callback,
+                } => {
+                    let current_block_hash = self.client.info().best_hash;
+
+                    let bsps_volunteered = self
+                        .client
+                        .runtime_api()
+                        .query_bsps_volunteered_for_file(current_block_hash, file_key)
+                        .unwrap_or_else(|_| Err(QueryBspsVolunteeredForFileError::InternalError));
+
+                    let volunteered = bsps_volunteered.map(|bsps| bsps.contains(&bsp_id));
+
+                    match callback.send(volunteered) {
+                        Ok(_) => {
+                            trace!(target: LOG_TARGET, "BSP volunteered status sent successfully");
+                        }
+                        Err(e) => {
+                            error!(target: LOG_TARGET, "Failed to send BSP volunteered status: {:?}", e);
+                        }
+                    }
+                }
                 BlockchainServiceCommand::QueryProviderMultiaddresses {
                     provider_id,
                     callback,

client/src/tasks/bsp_submit_proof.rs

@@ -302,9 +302,9 @@ where
             // - If the proof is outdated.
             // - If the Forest root of the BSP has changed.
             Box::pin(Self::should_retry_submit_proof(
-                cloned_sh_handler.clone(),
-                cloned_event.clone(),
-                cloned_forest_root.clone(),
+                cloned_sh_handler,
+                cloned_event,
+                cloned_forest_root,
                 error,
             )) as Pin<Box<dyn Future<Output = bool> + Send>>
         };

client/src/tasks/bsp_upload_file.rs

@@ -1,6 +1,9 @@
 use std::{
     collections::{HashMap, HashSet},
+    future::Future,
+    pin::Pin,
     str::FromStr,
+    sync::Arc,
     time::Duration,
 };
 
@@ -16,13 +19,13 @@ use shc_blockchain_service::{
     capacity_manager::CapacityRequestData,
     commands::{BlockchainServiceCommandInterface, BlockchainServiceCommandInterfaceExt},
     events::{NewStorageRequest, ProcessConfirmStoringRequest},
-    types::{ConfirmStoringRequest, RetryStrategy, SendExtrinsicOptions},
+    types::{ConfirmStoringRequest, RetryStrategy, SendExtrinsicOptions, WatchTransactionError},
 };
 use shc_common::{
     consts::CURRENT_FOREST_KEY,
     traits::StorageEnableRuntime,
     types::{
-        FileKey, FileKeyWithProof, FileMetadata, HashT, StorageProofsMerkleTrieLayout,
+        FileKey, FileKeyWithProof, FileMetadata, HashT, ProviderId, StorageProofsMerkleTrieLayout,
         StorageProviderId, BATCH_CHUNK_FILE_TRANSFER_MAX_SIZE,
     },
 };
@@ -76,7 +79,7 @@ pub struct BspUploadFileTask<NT, Runtime>
 where
     NT: ShNodeType<Runtime>,
     NT::FSH: BspForestStorageHandlerT<Runtime>,
-    Runtime: StorageEnableRuntime,
+    Runtime: StorageEnableRuntime + 'static,
 {
     storage_hub_handler: StorageHubHandler<NT, Runtime>,
     file_key_cleanup: Option<Runtime::Hash>,
@@ -101,7 +104,7 @@ where
 
 impl<NT, Runtime> BspUploadFileTask<NT, Runtime>
 where
-    NT: ShNodeType<Runtime>,
+    NT: ShNodeType<Runtime> + 'static,
     NT::FSH: BspForestStorageHandlerT<Runtime>,
     Runtime: StorageEnableRuntime,
 {
@@ -428,8 +431,8 @@ where
 
 impl<NT, Runtime> BspUploadFileTask<NT, Runtime>
 where
-    NT: ShNodeType<Runtime>,
-    NT::FSH: BspForestStorageHandlerT<Runtime>,
+    NT: ShNodeType<Runtime> + 'static,
+    NT::FSH: BspForestStorageHandlerT<Runtime> + 'static,
     Runtime: StorageEnableRuntime,
 {
     async fn handle_new_storage_request_event(
@@ -671,12 +674,36 @@ where
         }
         .into();
 
-        // Send extrinsic and wait for it to be included in the block.
-        let result = self
+        // Clone necessary data for the retry check.
+        let cloned_sh_handler = Arc::new(self.storage_hub_handler.clone());
+        let cloned_own_bsp_id = Arc::new(own_bsp_id.clone());
+        let cloned_file_key: Arc<Runtime::Hash> = Arc::new(file_key.clone().into());
+
+        let should_retry = move |error| {
+            let cloned_sh_handler = Arc::clone(&cloned_sh_handler);
+            let cloned_own_bsp_id = Arc::clone(&cloned_own_bsp_id);
+            let cloned_file_key = Arc::clone(&cloned_file_key);
+
+            // Check:
+            // - If we've already successfully volunteered for the file.
+            // - If the storage request is no longer open to volunteers.
+            // Also waits for the tick to be able to volunteer for the file has actually been reached,
+            // not the tick before the BSP can volunteer for the file. To make sure the chain wasn't
+            // spammed just before the BSP could volunteer for the file.
+            Box::pin(Self::should_retry_volunteer(
+                cloned_sh_handler,
+                cloned_own_bsp_id,
+                cloned_file_key,
+                error,
+            )) as Pin<Box<dyn Future<Output = bool> + Send>>
+        };
+
+        // Try to send the volunteer extrinsic
+        if let Err(e) = self
             .storage_hub_handler
             .blockchain
-            .send_extrinsic(
-                call.clone().into(),
+            .submit_extrinsic_with_retry(
+                call.clone(),
                 SendExtrinsicOptions::new(
                     Duration::from_secs(
                         self.storage_hub_handler
@@ -687,57 +714,43 @@ where
                     Some("fileSystem".to_string()),
                     Some("bspVolunteer".to_string()),
                 ),
+                RetryStrategy::default()
+                    .with_max_retries(self.config.max_try_count)
+                    .with_max_tip(self.config.max_tip.saturated_into())
+                    .with_should_retry(Some(Box::new(should_retry))),
+                false,
             )
-            .await?
-            .watch_for_success(&self.storage_hub_handler.blockchain)
-            .await;
+            .await
+        {
+            error!(target: LOG_TARGET, "Failed to volunteer for file {:x}: {:?}", file_key, e);
+        }
 
-        if let Err(e) = result {
+        // Check if the BSP has been registered as a volunteer for the file.
+        let volunteer_result = self
+            .storage_hub_handler
+            .blockchain
+            .query_bsp_volunteered_for_file(own_bsp_id, file_key.into())
+            .await
+            .map_err(|e| anyhow!("Failed to query BSP volunteered for file: {:?}", e))?;
+
+        // Handle the volunteer result.
+        if volunteer_result {
+            info!(
+                target: LOG_TARGET,
+                "🍾 BSP successfully volunteered for file {:x}",
+                file_key
+            );
+        } else {
             error!(
                 target: LOG_TARGET,
-                "Failed to volunteer for file {:?}: {:?}",
-                file_key,
-                e
+                "BSP not registered as a volunteer for file {:x}",
+                file_key
             );
-
-            // If the initial call errored out, it could mean the chain was spammed so the tick did not advance.
-            // Wait until the actual earliest volunteer tick to occur and retry volunteering.
-            self.storage_hub_handler
-                .blockchain
-                .wait_for_tick(earliest_volunteer_tick)
-                .await?;
-
-            // Send extrinsic and wait for it to be included in the block.
-            let result = self
-                .storage_hub_handler
-                .blockchain
-                .send_extrinsic(
-                    call,
-                    SendExtrinsicOptions::new(
-                        Duration::from_secs(
-                            self.storage_hub_handler
-                                .provider_config
-                                .blockchain_service
-                                .extrinsic_retry_timeout,
-                        ),
-                        Some("fileSystem".to_string()),
-                        Some("bspVolunteer".to_string()),
-                    ),
-                )
-                .await?
-                .watch_for_success(&self.storage_hub_handler.blockchain)
-                .await;
-
-            if let Err(e) = result {
-                error!(
-                    target: LOG_TARGET,
-                    "Failed to volunteer for file {:?} after retry in volunteer tick: {:?}",
-                    file_key,
-                    e
-                );
-
-                self.unvolunteer_file(file_key.into()).await;
-            }
+            self.unvolunteer_file(file_key.into()).await;
+            return Err(anyhow!(
+                "BSP not registered as a volunteer for file {:x}",
+                file_key
+            ));
         }
 
         Ok(())
@@ -750,6 +763,8 @@ where
         &mut self,
         event: RemoteUploadRequest<Runtime>,
     ) -> anyhow::Result<bool> {
+        debug!(target: LOG_TARGET, "Handling remote upload request for file key {:x}", event.file_key);
+
         let file_key = event.file_key.into();
         let mut write_file_storage = self.storage_hub_handler.file_storage.write().await;
 
@@ -997,6 +1012,85 @@ where
         return Ok(true);
     }
 
+    /// Function to determine if a volunteer request should be retried,
+    /// sending the same request again.
+    ///
+    /// This function will return `true` if and only if the following conditions are met:
+    /// 1. If the storage request is no longer open to volunteers.
+    /// 2. If we've already successfully volunteered for the file.
+    ///
+    /// Also waits for the tick to be able to volunteer for the file has actually been reached,
+    /// not the tick before the BSP can volunteer for the file. To make sure the chain wasn't
+    /// spammed just before the BSP could volunteer for the file.
+    async fn should_retry_volunteer(
+        sh_handler: Arc<StorageHubHandler<NT, Runtime>>,
+        bsp_id: Arc<ProviderId<Runtime>>,
+        file_key: Arc<Runtime::Hash>,
+        _error: WatchTransactionError,
+    ) -> bool {
+        // Wait for the tick to be able to volunteer for the file has actually been reached.
+        let earliest_volunteer_tick = match sh_handler
+            .blockchain
+            .query_file_earliest_volunteer_tick(*bsp_id, *file_key)
+            .await
+        {
+            Ok(tick) => tick,
+            Err(e) => {
+                error!(target: LOG_TARGET, "Failed to query file earliest volunteer block: {:?}", e);
+                return false;
+            }
+        };
+        match sh_handler
+            .blockchain
+            .wait_for_tick(earliest_volunteer_tick)
+            .await
+        {
+            Ok(_) => {}
+            Err(e) => {
+                error!(target: LOG_TARGET, "Failed to wait for tick: {:?}", e);
+                return false;
+            }
+        }
+
+        // Check if the storage request is no longer open to volunteers.
+        let can_volunteer = match sh_handler
+            .blockchain
+            .is_storage_request_open_to_volunteers(*file_key)
+            .await
+        {
+            Ok(can_volunteer) => can_volunteer,
+            Err(e) => {
+                error!(target: LOG_TARGET, "Failed to query file can volunteer: {:?}", e);
+                return false;
+            }
+        };
+
+        if !can_volunteer {
+            warn!(target: LOG_TARGET, "Storage request is no longer open to volunteers. Stop retrying.");
+            return false;
+        }
+
+        // Check if we've already successfully volunteered for the file.
+        let volunteered = match sh_handler
+            .blockchain
+            .query_bsp_volunteered_for_file(*bsp_id, *file_key)
+            .await
+        {
+            Ok(volunteered) => volunteered,
+            Err(e) => {
+                error!(target: LOG_TARGET, "Failed to query file volunteered: {:?}", e);
+                return false;
+            }
+        };
+
+        if volunteered {
+            info!(target: LOG_TARGET, "Already successfully volunteered for the file. Stop retrying.");
+            return false;
+        }
+
+        return true;
+    }
+
     async fn unvolunteer_file(&self, file_key: Runtime::Hash) {
         warn!(target: LOG_TARGET, "Unvolunteering file {:?}", file_key);