Feat: MSP move bucket (Moonsong-Labs#276)

* implement file transfer bucket allowlist; move bucket bsp tasks;

* add msp task

* add move-bucket test; fixes; drive-by refactor of waits

* fix lint/merge

* increase test time

* fix indexer payment tracking

* fix lint

* add todos

* ammed PR review

* fix merge

* fix merge

* fix merge

* ammend review comments

* Update client/file-transfer-service/src/handler.rs

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

* fix warning from merge

---------

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

Author: links234

Cargo.lock

@@ -1,7 +1,10 @@
 use anyhow::Result;
 use async_trait::async_trait;
 use log::{debug, warn};
+use sc_network::Multiaddr;
 use serde_json::Number;
+use sp_api::ApiError;
+use sp_core::H256;
 
 use pallet_file_system_runtime_api::{
     QueryBspConfirmChunksToProveForFileError, QueryFileEarliestVolunteerTickError,
@@ -17,11 +20,9 @@ use pallet_storage_providers_runtime_api::{
 };
 use shc_actors_framework::actor::ActorHandle;
 use shc_common::types::{
-    BlockNumber, BucketId, ChunkId, ForestLeaf, MainStorageProviderId, Multiaddresses, ProviderId,
+    BlockNumber, BucketId, ChunkId, ForestLeaf, MainStorageProviderId, ProviderId,
     RandomnessOutput, StorageHubEventsVec, StorageProviderId, TickNumber, TrieRemoveMutation,
 };
-use sp_api::ApiError;
-use sp_core::H256;
 use storage_hub_runtime::{AccountId, Balance, StorageDataUnit};
 
 use super::{
@@ -92,7 +93,7 @@ pub enum BlockchainServiceCommand {
     QueryProviderMultiaddresses {
         provider_id: ProviderId,
         callback:
-            tokio::sync::oneshot::Sender<Result<Multiaddresses, QueryProviderMultiaddressesError>>,
+            tokio::sync::oneshot::Sender<Result<Vec<Multiaddr>, QueryProviderMultiaddressesError>>,
     },
     QueueSubmitProofRequest {
         request: SubmitProofRequest,
@@ -248,8 +249,8 @@ pub trait BlockchainServiceInterface {
     /// Query the MSP multiaddresses.
     async fn query_provider_multiaddresses(
         &self,
-        msp_id: ProviderId,
-    ) -> Result<Multiaddresses, QueryProviderMultiaddressesError>;
+        provider_id: ProviderId,
+    ) -> Result<Vec<Multiaddr>, QueryProviderMultiaddressesError>;
 
     /// Queue a SubmitProofRequest to be processed.
     async fn queue_submit_proof_request(&self, request: SubmitProofRequest) -> Result<()>;
@@ -517,7 +518,7 @@ impl BlockchainServiceInterface for ActorHandle<BlockchainService> {
     async fn query_provider_multiaddresses(
         &self,
         provider_id: ProviderId,
-    ) -> Result<Multiaddresses, QueryProviderMultiaddressesError> {
+    ) -> Result<Vec<Multiaddr>, QueryProviderMultiaddressesError> {
         let (callback, rx) = tokio::sync::oneshot::channel();
         let message = BlockchainServiceCommand::QueryProviderMultiaddresses {
             provider_id,

client/blockchain-service/src/commands.rs

@@ -246,6 +246,39 @@ pub struct SpStopStoringInsolventUser {
 }
 impl EventBusMessage for SpStopStoringInsolventUser {}
 
+#[derive(Debug, Clone)]
+pub struct MoveBucketRequested {
+    pub bucket_id: BucketId,
+    pub new_msp_id: ProviderId,
+}
+impl EventBusMessage for MoveBucketRequested {}
+
+#[derive(Debug, Clone)]
+pub struct MoveBucketRequestedForNewMsp {
+    pub bucket_id: BucketId,
+}
+impl EventBusMessage for MoveBucketRequestedForNewMsp {}
+
+#[derive(Debug, Clone)]
+pub struct MoveBucketRejected {
+    pub bucket_id: BucketId,
+    pub msp_id: ProviderId,
+}
+impl EventBusMessage for MoveBucketRejected {}
+
+#[derive(Debug, Clone)]
+pub struct MoveBucketAccepted {
+    pub bucket_id: BucketId,
+    pub msp_id: ProviderId,
+}
+impl EventBusMessage for MoveBucketAccepted {}
+
+#[derive(Debug, Clone)]
+pub struct MoveBucketExpired {
+    pub bucket_id: BucketId,
+    pub msp_id: ProviderId,
+}
+impl EventBusMessage for MoveBucketExpired {}
 /// BSP stopped storing a specific file.
 ///
 /// This event is emitted when a BSP confirm stop storing a file.
@@ -300,6 +333,11 @@ pub struct BlockchainServiceEventBusProvider {
     user_without_funds_event_bus: EventBus<UserWithoutFunds>,
     sp_stop_storing_insolvent_user_event_bus: EventBus<SpStopStoringInsolventUser>,
     finalised_msp_stopped_storing_bucket_event_bus: EventBus<FinalisedMspStoppedStoringBucket>,
+    move_bucket_requested_event_bus: EventBus<MoveBucketRequested>,
+    move_bucket_rejected_event_bus: EventBus<MoveBucketRejected>,
+    move_bucket_accepted_event_bus: EventBus<MoveBucketAccepted>,
+    move_bucket_expired_event_bus: EventBus<MoveBucketExpired>,
+    move_bucket_requested_for_new_msp_event_bus: EventBus<MoveBucketRequestedForNewMsp>,
     bsp_stop_storing_event_bus: EventBus<BspConfirmStoppedStoring>,
     finalised_bsp_stop_storing_event_bus: EventBus<FinalisedBspConfirmStoppedStoring>,
     notify_period_event_bus: EventBus<NotifyPeriod>,
@@ -323,6 +361,11 @@ impl BlockchainServiceEventBusProvider {
             user_without_funds_event_bus: EventBus::new(),
             sp_stop_storing_insolvent_user_event_bus: EventBus::new(),
             finalised_msp_stopped_storing_bucket_event_bus: EventBus::new(),
+            move_bucket_requested_event_bus: EventBus::new(),
+            move_bucket_rejected_event_bus: EventBus::new(),
+            move_bucket_accepted_event_bus: EventBus::new(),
+            move_bucket_expired_event_bus: EventBus::new(),
+            move_bucket_requested_for_new_msp_event_bus: EventBus::new(),
             bsp_stop_storing_event_bus: EventBus::new(),
             finalised_bsp_stop_storing_event_bus: EventBus::new(),
             notify_period_event_bus: EventBus::new(),
@@ -422,6 +465,36 @@ impl ProvidesEventBus<FinalisedMspStoppedStoringBucket> for BlockchainServiceEve
     }
 }
 
+impl ProvidesEventBus<MoveBucketRequested> for BlockchainServiceEventBusProvider {
+    fn event_bus(&self) -> &EventBus<MoveBucketRequested> {
+        &self.move_bucket_requested_event_bus
+    }
+}
+
+impl ProvidesEventBus<MoveBucketRejected> for BlockchainServiceEventBusProvider {
+    fn event_bus(&self) -> &EventBus<MoveBucketRejected> {
+        &self.move_bucket_rejected_event_bus
+    }
+}
+
+impl ProvidesEventBus<MoveBucketAccepted> for BlockchainServiceEventBusProvider {
+    fn event_bus(&self) -> &EventBus<MoveBucketAccepted> {
+        &self.move_bucket_accepted_event_bus
+    }
+}
+
+impl ProvidesEventBus<MoveBucketExpired> for BlockchainServiceEventBusProvider {
+    fn event_bus(&self) -> &EventBus<MoveBucketExpired> {
+        &self.move_bucket_expired_event_bus
+    }
+}
+
+impl ProvidesEventBus<MoveBucketRequestedForNewMsp> for BlockchainServiceEventBusProvider {
+    fn event_bus(&self) -> &EventBus<MoveBucketRequestedForNewMsp> {
+        &self.move_bucket_requested_for_new_msp_event_bus
+    }
+}
+
 impl ProvidesEventBus<BspConfirmStoppedStoring> for BlockchainServiceEventBusProvider {
     fn event_bus(&self) -> &EventBus<BspConfirmStoppedStoring> {
         &self.bsp_stop_storing_event_bus

client/blockchain-service/src/events.rs

@@ -52,8 +52,9 @@ use crate::{
     events::{
         AcceptedBspVolunteer, BlockchainServiceEventBusProvider, BspConfirmStoppedStoring,
         FinalisedBspConfirmStoppedStoring, FinalisedMspStoppedStoringBucket,
-        FinalisedTrieRemoveMutationsApplied, LastChargeableInfoUpdated, NewStorageRequest,
-        SlashableProvider, SpStopStoringInsolventUser, UserWithoutFunds,
+        FinalisedTrieRemoveMutationsApplied, LastChargeableInfoUpdated, MoveBucketAccepted,
+        MoveBucketExpired, MoveBucketRejected, MoveBucketRequested, MoveBucketRequestedForNewMsp,
+        NewStorageRequest, SlashableProvider, SpStopStoringInsolventUser, UserWithoutFunds,
     },
     state::{
         BlockchainServiceStateStore, LastProcessedBlockNumberCf,
@@ -504,7 +505,8 @@ impl Actor for BlockchainService {
                         .unwrap_or_else(|_| {
                             error!(target: LOG_TARGET, "Failed to query provider multiaddresses");
                             Err(QueryProviderMultiaddressesError::InternalError)
-                        });
+                        })
+                        .map(convert_raw_multiaddresses_to_multiaddr);
 
                     match callback.send(multiaddresses) {
                         Ok(_) => {
@@ -1248,6 +1250,39 @@ impl BlockchainService {
                                 size,
                             })
                         }
+                        RuntimeEvent::FileSystem(
+                            pallet_file_system::Event::MoveBucketRequested {
+                                who: _,
+                                bucket_id,
+                                new_msp_id,
+                            },
+                        ) => {
+                            self.emit(MoveBucketRequested {
+                                bucket_id,
+                                new_msp_id,
+                            });
+                            if self.provider_ids.contains(&new_msp_id) {
+                                self.emit(MoveBucketRequestedForNewMsp { bucket_id });
+                            }
+                        }
+                        RuntimeEvent::FileSystem(
+                            pallet_file_system::Event::MoveBucketRejected { bucket_id, msp_id },
+                        ) => {
+                            self.emit(MoveBucketRejected { bucket_id, msp_id });
+                        }
+                        RuntimeEvent::FileSystem(
+                            pallet_file_system::Event::MoveBucketAccepted { bucket_id, msp_id },
+                        ) => {
+                            self.emit(MoveBucketAccepted { bucket_id, msp_id });
+                        }
+                        RuntimeEvent::FileSystem(
+                            pallet_file_system::Event::MoveBucketRequestExpired {
+                                bucket_id,
+                                msp_id,
+                            },
+                        ) => {
+                            self.emit(MoveBucketExpired { bucket_id, msp_id });
+                        }
                         RuntimeEvent::FileSystem(
                             pallet_file_system::Event::BspConfirmStoppedStoring {
                                 bsp_id,

client/blockchain-service/src/handler.rs

@@ -57,21 +57,25 @@ pub fn get_events_at_block(
 pub fn convert_raw_multiaddresses_to_multiaddr(multiaddresses: Multiaddresses) -> Vec<Multiaddr> {
     let mut multiaddress_vec: Vec<Multiaddr> = Vec::new();
     for raw_multiaddr in multiaddresses.into_iter() {
-        let multiaddress = match std::str::from_utf8(&raw_multiaddr) {
-            Ok(s) => match Multiaddr::from_str(s) {
-                Ok(multiaddr) => multiaddr,
-                Err(e) => {
-                    error!("Failed to parse Multiaddress from string: {:?}", e);
-                    continue;
-                }
-            },
+        if let Some(multiaddress) = convert_raw_multiaddress_to_multiaddr(&raw_multiaddr) {
+            multiaddress_vec.push(multiaddress);
+        }
+    }
+    multiaddress_vec
+}
+
+pub fn convert_raw_multiaddress_to_multiaddr(raw_multiaddr: &[u8]) -> Option<Multiaddr> {
+    match std::str::from_utf8(raw_multiaddr) {
+        Ok(s) => match Multiaddr::from_str(s) {
+            Ok(multiaddr) => Some(multiaddr),
             Err(e) => {
-                error!("Failed to parse Multiaddress from bytes: {:?}", e);
-                continue;
+                error!("Failed to parse Multiaddress from string: {:?}", e);
+                None
             }
-        };
-
-        multiaddress_vec.push(multiaddress);
+        },
+        Err(e) => {
+            error!("Failed to parse Multiaddress from bytes: {:?}", e);
+            None
+        }
     }
-    multiaddress_vec
 }

client/common/src/blockchain_utils.rs

@@ -49,6 +49,7 @@ pub type StorageProviderId = pallet_storage_providers::types::StorageProviderId<
 pub type MainStorageProviderId = pallet_storage_providers::types::ProviderIdFor<Runtime>;
 pub type ProviderId = pallet_proofs_dealer::types::ProviderIdFor<Runtime>;
 pub type Multiaddresses = pallet_storage_providers::types::Multiaddresses<Runtime>;
+pub type MultiAddress = pallet_storage_providers::types::MultiAddress<Runtime>;
 pub type RandomnessOutput = pallet_proofs_dealer::types::RandomnessOutputFor<Runtime>;
 pub type ForestLeaf = pallet_proofs_dealer::types::KeyFor<Runtime>;
 pub type ForestRoot = pallet_proofs_dealer::types::ForestRootFor<Runtime>;
@@ -62,6 +63,8 @@ pub type KeyProofs = BTreeMap<ForestLeaf, KeyProof>;
 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>;
+
 /// Type alias for the events vector.
 ///
 /// The events vector is a storage element in the FRAME system pallet, which stores all the events

client/common/src/types.rs

@@ -25,6 +25,7 @@ async-trait = { workspace = true }
 array-bytes = { workspace = true }
 futures = { workspace = true }
 codec = { workspace = true }
+chrono = { workspace = true }
 
 # Substrate
 sc-client-api = { workspace = true }