feat: msp check previous block

client/blockchain-service/src/commands.rs

@@ -64,6 +64,10 @@ pub enum BlockchainServiceCommand {
         block_number: BlockNumber,
         callback: tokio::sync::oneshot::Sender<tokio::sync::oneshot::Receiver<()>>,
     },
+    WaitForNumBlocks {
+        number_of_blocks: BlockNumber,
+        callback: tokio::sync::oneshot::Sender<tokio::sync::oneshot::Receiver<()>>,
+    },
     WaitForTick {
         tick_number: TickNumber,
         callback:
@@ -239,6 +243,9 @@ pub trait BlockchainServiceInterface {
     /// Wait for a block number.
     async fn wait_for_block(&self, block_number: BlockNumber) -> Result<()>;
 
+    /// Wait for a number blocks to pass.
+    async fn wait_for_num_blocks(&self, number_of_blocks: BlockNumber) -> Result<()>;
+
     /// Wait for a tick number.
     async fn wait_for_tick(&self, tick_number: TickNumber) -> Result<(), ApiError>;
 
@@ -487,6 +494,19 @@ where
         Ok(())
     }
 
+    async fn wait_for_num_blocks(&self, number_of_blocks: BlockNumber) -> Result<()> {
+        let (callback, rx) = tokio::sync::oneshot::channel();
+        // Build command to send to blockchain service.
+        let message = BlockchainServiceCommand::WaitForNumBlocks {
+            number_of_blocks,
+            callback,
+        };
+        self.send(message).await;
+        let rx = rx.await.expect("Failed to receive response from BlockchainService. Probably means BlockchainService has crashed.");
+        rx.await.expect("Failed to wait for block");
+        Ok(())
+    }
+
     async fn wait_for_tick(&self, tick_number: TickNumber) -> Result<(), ApiError> {
         let (callback, rx) = tokio::sync::oneshot::channel();
         // Build command to send to blockchain service.

client/blockchain-service/src/handler.rs

@@ -31,7 +31,10 @@
 use shc_actors_framework::actor::{Actor, ActorEventLoop};
 use shc_common::{
     blockchain_utils::{convert_raw_multiaddresses_to_multiaddr, get_events_at_block},
-    types::{BlockNumber, ParachainClient, TickNumber},
+    types::{
+        BlockNumber, EitherBucketOrBspId, Fingerprint, ParachainClient, StorageProviderId,
+        StorageRequestMetadata, TickNumber, BCSV_KEY_TYPE,
+    },
 };
 
 use crate::{
@@ -352,6 +355,28 @@
                         }
                     }
                 }
+                BlockchainServiceCommand::WaitForNumBlocks {
+                    number_of_blocks,
+                    callback,
+                } => {
+                    let current_block_number = self.client.info().best_number;
+
+                    let (tx, rx) = tokio::sync::oneshot::channel();
+
+                    self.wait_for_block_request_by_number
+                        .entry(current_block_number + number_of_blocks)
+                        .or_insert_with(Vec::new)
+                        .push(tx);
+
+                    match callback.send(rx) {
+                        Ok(_) => {
+                            trace!(target: LOG_TARGET, "Block message receiver sent successfully");
+                        }
+                        Err(e) => {
+                            error!(target: LOG_TARGET, "Failed to send block message receiver: {:?}", e);
+                        }
+                    }
+                }
                 BlockchainServiceCommand::WaitForTick {
                     tick_number,
                     callback,
@@ -1268,7 +1293,44 @@
             Some(ManagedProvider::Bsp(_)) => {
                 self.bsp_initial_sync();
             }
-            Some(ManagedProvider::Msp(_)) => {
+            Some(StorageProviderId::MainStorageProvider(msp_id)) => {
+                // TODO: Send events to check that this node has a Forest Storage for each Bucket this MSP manages.
+                // TODO: Catch up to Forest root writes in the Bucket's Forests.
+
+                info!(target: LOG_TARGET, "Checking for storage requests for this MSP");
+
+                let storage_requests: BTreeMap<H256, StorageRequestMetadata> = match self
+                    .client
+                    .runtime_api()
+                    .pending_storage_requests_by_msp(block_hash, msp_id)
+                {
+                    Ok(sr) => sr,
+                    Err(_) => {
+                        // If querying for pending storage requests fail, do not try to answer them
+                        warn!(target: LOG_TARGET, "Failed to get pending storage request");
+                        return;
+                    }
+                };
+
+                info!(
+                    "We have {} pending storage requests",
+                    storage_requests.len()
+                );
+
+                // loop over each pending storage requests to start a new storage request task for the MSP
+                for (file_key, sr) in storage_requests {
+                    self.emit(NewStorageRequest {
+                        who: sr.owner,
+                        file_key: file_key.into(),
+                        bucket_id: sr.bucket_id,
+                        location: sr.location,
+                        fingerprint: Fingerprint::from(sr.fingerprint.as_bytes()),
+                        size: sr.size,
+                        user_peer_ids: sr.user_peer_ids,
+                        expires_at: sr.expires_at,
+                    })
+                }
+
                 self.msp_initial_sync();
             }
             None => {

client/common/src/types.rs

@@ -76,6 +76,7 @@ pub type Balance = pallet_storage_providers::types::BalanceOf<Runtime>;
 pub type OpaqueBlock = storage_hub_runtime::opaque::Block;
 pub type BlockHash = <OpaqueBlock as BlockT>::Hash;
 pub type PeerId = pallet_file_system::types::PeerId<Runtime>;
+pub type StorageRequestMetadata = pallet_file_system::types::StorageRequestMetadata<Runtime>;
 pub type MaxBatchConfirmStorageRequests =
     <Runtime as pallet_file_system::Config>::MaxBatchConfirmStorageRequests;
 

client/rpc/src/lib.rs

@@ -22,15 +22,7 @@ use tokio::{fs, fs::create_dir_all, sync::RwLock};
 
 use pallet_file_system_runtime_api::FileSystemApi as FileSystemRuntimeApi;
 use pallet_proofs_dealer_runtime_api::ProofsDealerApi as ProofsDealerRuntimeApi;
-use shc_common::{
-    consts::CURRENT_FOREST_KEY,
-    types::{
-        BackupStorageProviderId, BlockNumber, BucketId, ChunkId, CustomChallenge, FileMetadata,
-        ForestLeaf, HashT, KeyProof, KeyProofs, MainStorageProviderId, ProofsDealerProviderId,
-        Proven, RandomnessOutput, StorageProof, StorageProofsMerkleTrieLayout, BCSV_KEY_TYPE,
-        FILE_CHUNK_SIZE,
-    },
-};
+use shc_common::{consts::CURRENT_FOREST_KEY, types::*};
 use shc_file_manager::traits::{ExcludeType, FileDataTrie, FileStorage, FileStorageError};
 use shc_forest_manager::traits::{ForestStorage, ForestStorageHandler};
 use sp_core::{sr25519::Pair as Sr25519Pair, Encode, Pair, H256};
@@ -323,6 +315,7 @@ where
             BlockNumber,
             ChunkId,
             BucketId,
+            StorageRequestMetadata,
         >,
     FL: FileStorage<StorageProofsMerkleTrieLayout> + Send + Sync,
     FSH: ForestStorageHandler + Send + Sync + 'static,

node/src/rpc.rs

@@ -15,10 +15,7 @@ use sc_consensus_manual_seal::{
 };
 use sc_rpc::DenyUnsafe;
 use sc_transaction_pool_api::TransactionPool;
-use shc_common::types::{
-    BackupStorageProviderId, BlockNumber, BucketId, ChunkId, CustomChallenge, ForestLeaf,
-    MainStorageProviderId, ProofsDealerProviderId, RandomnessOutput,
-};
+use shc_common::types::*;
 use shc_forest_manager::traits::ForestStorageHandler;
 use shc_rpc::{StorageHubClientApiServer, StorageHubClientRpc, StorageHubClientRpcConfig};
 use sp_api::ProvideRuntimeApi;
@@ -73,6 +70,7 @@ where
             BlockNumber,
             ChunkId,
             BucketId,
+            StorageRequestMetadata,
         >,
     P: TransactionPool + Send + Sync + 'static,
     FL: FileStorageT,

node/src/tasks/bsp_charge_fees.rs

@@ -272,6 +272,9 @@ where
             .map_err(|e| anyhow!("Failed to get metadata from Forest: {:?}", e))?;
 
         if !user_files.is_empty() {
+            // We only take the first file of the list in order to generate a proof submit it with an extrinsic and then release the lock to process the next file and generate the next proof.
+            // It is not ideal because it means one extrinsic per file but batch deletion is not yet implemented.
+            // TODO: Improve it once batch deletion is implemented.
             let (file_key, metadata) = user_files.first().expect("User files is not empty");
             let bucket_id = H256::from_slice(metadata.bucket_id().as_ref());
             let location = sp_runtime::BoundedVec::truncate_from(metadata.location().clone());

node/src/tasks/user_sends_file.rs

@@ -317,7 +317,7 @@ where
                         {
                             warn!(
                                 target: LOG_TARGET,
-                                "Batch upload rejected by peer {:?}, retrying... (attempt {})",
+                                "Final batch upload rejected by peer {:?}, retrying... (attempt {})",
                                 peer_id,
                                 retry_attempts + 1
                             );
@@ -326,7 +326,27 @@ where
                             // Wait for a short time before retrying
                             tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                         }
-                        Err(RequestError::RequestFailure(RequestFailure::Refused)) => {
+                        Err(RequestError::RequestFailure(RequestFailure::Network(_)))
+                        | Err(RequestError::RequestFailure(RequestFailure::NotConnected))
+                            if retry_attempts < 10 =>
+                        {
+                            warn!(
+                                target: LOG_TARGET,
+                                "Unable to upload final batch to peer {:?}, retrying... (attempt {})",
+                                peer_id,
+                                retry_attempts + 1
+                            );
+                            retry_attempts += 1;
+
+                            // Wait a bit for the MSP to be online
+                            self.storage_hub_handler
+                                .blockchain
+                                .wait_for_num_blocks(5)
+                                .await?;
+                        }
+                        Err(RequestError::RequestFailure(RequestFailure::Refused))
+                        | Err(RequestError::RequestFailure(RequestFailure::Network(_)))
+                        | Err(RequestError::RequestFailure(RequestFailure::NotConnected)) => {
                             // Return an error if the provider refused to answer.
                             return Err(anyhow::anyhow!("Failed to send file {:?}", file_key));
                         }
@@ -388,7 +408,7 @@ where
                         .upload_request(peer_id, file_key.as_ref().into(), proof.clone(), None)
                         .await;
 
-                    match upload_response {
+                    match upload_response.as_ref() {
                         Ok(r) => {
                             debug!(
                                 target: LOG_TARGET,
@@ -421,7 +441,27 @@ where
                             // Wait for a short time before retrying
                             tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                         }
-                        Err(RequestError::RequestFailure(RequestFailure::Refused)) => {
+                        Err(RequestError::RequestFailure(RequestFailure::Network(_)))
+                        | Err(RequestError::RequestFailure(RequestFailure::NotConnected))
+                            if retry_attempts < 10 =>
+                        {
+                            warn!(
+                                target: LOG_TARGET,
+                                "Unable to upload final batch to peer {:?}, retrying... (attempt {})",
+                                peer_id,
+                                retry_attempts + 1
+                            );
+                            retry_attempts += 1;
+
+                            // Wait a bit for the MSP to be online
+                            self.storage_hub_handler
+                                .blockchain
+                                .wait_for_num_blocks(5)
+                                .await?;
+                        }
+                        Err(RequestError::RequestFailure(RequestFailure::Refused))
+                        | Err(RequestError::RequestFailure(RequestFailure::Network(_)))
+                        | Err(RequestError::RequestFailure(RequestFailure::NotConnected)) => {
                             // Return an error if the provider refused to answer.
                             return Err(anyhow::anyhow!("Failed to send file {:?}", file_key));
                         }

pallets/file-system/runtime-api/Cargo.toml

@@ -18,8 +18,10 @@ targets = ["x86_64-unknown-linux-gnu"]
 codec = { workspace = true, features = ["derive"] }
 scale-info = { workspace = true }
 sp-api = { workspace = true }
+sp-core = { workspace = true }
 sp-runtime = { workspace = true }
+sp-std = { workspace = true }
 
 [features]
 default = ["std"]
-std = ["codec/std", "sp-api/std", "sp-runtime/std"]
+std = ["codec/std", "sp-api/std", "sp-runtime/std", "sp-std/std"]

pallets/file-system/runtime-api/src/lib.rs

@@ -3,7 +3,9 @@
 use codec::{Codec, Decode, Encode};
 use scale_info::prelude::vec::Vec;
 use scale_info::TypeInfo;
+use sp_core::H256;
 use sp_runtime::RuntimeDebug;
+use sp_std::collections::btree_map::BTreeMap;
 
 /// Error type for the `is_storage_request_open_to_volunteers` runtime API call.
 #[derive(Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo)]
@@ -53,21 +55,29 @@ pub enum GenericApplyDeltaEventInfoError {
     DecodeError,
 }
 
+/// Error type for the `pending_storage_requests_by_msp`.
+#[derive(Eq, PartialEq, Encode, Decode, RuntimeDebug, TypeInfo)]
+pub enum StorageRequestsByMspError {
+    FailedToRetrieveStorageRequests,
+}
+
 sp_api::decl_runtime_apis! {
     #[api_version(1)]
-    pub trait FileSystemApi<BackupStorageProviderId, MainStorageProviderId, FileKey, TickNumber, ChunkId, GenericApplyDeltaEventInfo>
+    pub trait FileSystemApi<BackupStorageProviderId, MainStorageProviderId, FileKey, TickNumber, ChunkId, GenericApplyDeltaEventInfo, StorageRequestMetadata>
     where
         BackupStorageProviderId: Codec,
         MainStorageProviderId: Codec,
         FileKey: Codec,
         TickNumber: Codec,
         ChunkId: Codec,
         GenericApplyDeltaEventInfo: Codec,
+        StorageRequestMetadata: Codec,
     {
         fn is_storage_request_open_to_volunteers(file_key: FileKey) -> Result<bool, IsStorageRequestOpenToVolunteersError>;
         fn query_earliest_file_volunteer_tick(bsp_id: BackupStorageProviderId, file_key: FileKey) -> Result<TickNumber, QueryFileEarliestVolunteerTickError>;
         fn query_bsp_confirm_chunks_to_prove_for_file(bsp_id: BackupStorageProviderId, file_key: FileKey) -> Result<Vec<ChunkId>, QueryBspConfirmChunksToProveForFileError>;
         fn query_msp_confirm_chunks_to_prove_for_file(msp_id: MainStorageProviderId, file_key: FileKey) -> Result<Vec<ChunkId>, QueryMspConfirmChunksToProveForFileError>;
         fn decode_generic_apply_delta_event_info(encoded_event_info: Vec<u8>) -> Result<GenericApplyDeltaEventInfo, GenericApplyDeltaEventInfoError>;
+        fn pending_storage_requests_by_msp(msp_id: MainStorageProviderId) -> BTreeMap<H256, StorageRequestMetadata>;
     }
 }

pallets/file-system/src/utils.rs

@@ -34,6 +34,7 @@
     ReadBucketsInterface, ReadProvidersInterface, ReadStorageProvidersInterface,
     ReadUserSolvencyInterface, TrieAddMutation, TrieRemoveMutation,
 };
+use sp_std::collections::btree_map::BTreeMap;
 
 use crate::{
     pallet,
@@ -71,7 +72,7 @@
     }};
     // Handle boolean type
     ($condition:expr, $error_msg:expr, $error_type:path, bool) => {{
         if !$condition {
             #[cfg(test)]
             unreachable!($error_msg);
 
@@ -2841,6 +2842,22 @@
     ) -> Result<BucketIdFor<T>, codec::Error> {
         BucketIdFor::<T>::decode(&mut encoded_event_info)
     }
+
+    pub fn pending_storage_requests_by_msp(
+        msp_id: ProviderIdFor<T>,
+    ) -> BTreeMap<MerkleHash<T>, StorageRequestMetadata<T>> {
+        // Get the storage requests for a specific MSP
+        StorageRequests::<T>::iter()
+            .filter(|(_, metadata)| {
+                if let Some(msp) = metadata.msp {
+                    msp.0 == msp_id && !msp.1
+                } else {
+                    false
+                }
+            })
+            .map(|(file_key, metadata)| (file_key, metadata))
+            .collect()
+    }
 }
 
 mod hooks {