runtime: storage: Add extrinsic.

- add `invite_distribution_bucket_operator`

runtime-modules/storage/src/distribution_bucket_picker.rs

@@ -61,7 +61,7 @@ impl<T: Trait> DistributionBucketPicker<T> {
         bucket_number: u32,
         seed: Rc<RefCell<T::Hash>>, //     seed: RefCell<T::Hash>
     ) -> BTreeSet<T::DistributionBucketId> {
-        let mut working_ids = ids.clone();
+        let mut working_ids = ids;
         let mut result_ids = BTreeSet::default();
 
         for _ in 0..bucket_number {
@@ -87,7 +87,7 @@ impl<T: Trait> DistributionBucketPicker<T> {
             return Module::<T>::get_initial_random_seed();
         }
 
-        let current_seed = seed.borrow().clone();
+        let current_seed = *seed.borrow();
         seed.replace(T::Randomness::random(current_seed.as_ref()))
     }
 }

runtime-modules/storage/src/lib.rs

@@ -70,6 +70,8 @@
 //! updates "distributing" flag for the distribution bucket.
 //! - [update_families_in_dynamic_bag_creation_policy](./struct.Module.html#method.update_families_in_dynamic_bag_creation_policy) -
 //!  updates distribution bucket families used in given dynamic bag creation policy.
+//! - [invite_distribution_bucket_operator](./struct.Module.html#method.invite_distribution_bucket_operator) -
+//!  invites a distribution bucket operator.
 //!
 //! #### Public methods
 //! Public integration methods are exposed via the [DataObjectStorage](./trait.DataObjectStorage.html)
@@ -141,6 +143,10 @@ use storage_bucket_picker::StorageBucketPicker;
 // TODO: mut in extrinsics
 // TODO: dynamic bag creation - distributed buckets
 
+// TODO:
+// How to be sure that we don't have already accepted invitation? We need to have it in the bucket
+// a separate map. DistributionOperatorId is not sustainable - verify that on Monday.
+// TODO: test invite_distribution_bucket_operator with accepted invitation.
 /// Public interface for the storage module.
 pub trait DataObjectStorage<T: Trait> {
     /// Validates upload parameters and conditions (like global uploading block).
@@ -250,6 +256,17 @@ pub trait Trait: frame_system::Trait + balances::Trait + membership::Trait {
         + MaybeSerialize
         + PartialEq;
 
+    /// Distribution bucket operator ID type (relationship between distribution bucket and
+    /// distribution operator).
+    type DistributionBucketOperatorId: Parameter
+        + Member
+        + BaseArithmetic
+        + Codec
+        + Default
+        + Copy
+        + MaybeSerialize
+        + PartialEq;
+
     /// Defines max allowed storage bucket number.
     type MaxStorageBucketNumber: Get<u64>;
 
@@ -317,6 +334,10 @@ pub trait Trait: frame_system::Trait + balances::Trait + membership::Trait {
         origin: Self::Origin,
         worker_id: WorkerId<Self>,
     ) -> DispatchResult;
+
+    /// Validate distribution worker existence.
+    /// TODO: Refactor after merging with the Olympia release.
+    fn ensure_distribution_worker_exists(worker_id: &WorkerId<Self>) -> DispatchResult;
 }
 
 /// Operations with local pallet account.
@@ -773,27 +794,36 @@ impl<Balance: Saturating + Copy> BagChangeInfo<Balance> {
     }
 }
 
+/// Type alias for the DistributionBucketObject.
+pub type DistributionBucketFamily<T> =
+    DistributionBucketFamilyObject<<T as Trait>::DistributionBucketId, WorkerId<T>>;
+
 /// Distribution bucket family.
 #[cfg_attr(feature = "std", derive(Serialize, Deserialize))]
 #[derive(Encode, Decode, Default, Clone, PartialEq, Eq, Debug)]
-pub struct DistributionBucketFamily<DistributionBucketId: Ord> {
+pub struct DistributionBucketFamilyObject<DistributionBucketId: Ord, WorkerId: Ord> {
     /// Distribution bucket map.
-    pub distribution_buckets: BTreeMap<DistributionBucketId, DistributionBucket>,
+    pub distribution_buckets: BTreeMap<DistributionBucketId, DistributionBucketObject<WorkerId>>,
 }
 
+/// Type alias for the DistributionBucketObject.
+pub type DistributionBucket<T> = DistributionBucketObject<WorkerId<T>>;
+
 /// Distribution bucket.
 #[cfg_attr(feature = "std", derive(Serialize, Deserialize))]
 #[derive(Encode, Decode, Default, Clone, PartialEq, Eq, Debug)]
-pub struct DistributionBucket {
+pub struct DistributionBucketObject<WorkerId: Ord> {
     /// Distribution bucket accepts new bags.
     pub accepting_new_bags: bool,
 
     /// Distribution bucket serves objects.
     pub distributing: bool,
-    //TODO:
-    // pending_invitations: BTreeSet<WorkerId>,
-    // number_of_pending_data_objects: u32,
-    // number_of_operators: u32,
+
+    /// Pending invitations for workers to distribute the bucket.
+    pub pending_invitations: BTreeSet<WorkerId>,
+
+    /// Active operators to distribute the bucket.
+    pub operators: BTreeSet<WorkerId>,
 }
 
 decl_storage! {
@@ -852,7 +882,7 @@ decl_storage! {
         /// Distribution bucket families.
         pub DistributionBucketFamilyById get (fn distribution_bucket_family_by_id):
             map hasher(blake2_128_concat) T::DistributionBucketFamilyId =>
-            DistributionBucketFamily<T::DistributionBucketId>;
+            DistributionBucketFamily<T>;
 
         /// Total number of distribution bucket families in the system.
         pub DistributionBucketFamilyNumber get(fn distribution_bucket_family_number): u64;
@@ -1091,6 +1121,16 @@ decl_event! {
             DynamicBagType,
             BTreeMap<DistributionBucketFamilyId, u32>
         ),
+
+        /// Emits on dynamic bag creation policy update (distribution bucket families).
+        /// Params
+        /// - distribution bucket family ID
+        /// - distribution bucket ID
+        DistributionBucketOperatorInvited(
+            DistributionBucketFamilyId,
+            DistributionBucketId,
+            WorkerId,
+        ),
     }
 }
 
@@ -1240,6 +1280,15 @@ decl_error! {
 
         /// The new `DistributionBucketsPerBagLimit` number is too high.
         DistributionBucketsPerBagLimitTooHigh,
+
+        /// Distribution provider operator doesn't exist.
+        DistributionProviderOperatorDoesntExist,
+
+        /// Distribution provider operator already invited.
+        DistributionProviderOperatorAlreadyInvited,
+
+        /// Distribution provider operator already set.
+        DistributionProviderOperatorAlreadySet,
     }
 }
 
@@ -1833,9 +1882,7 @@ decl_module! {
 
             Self::increment_distribution_family_number();
 
-            let family = DistributionBucketFamily {
-                distribution_buckets: BTreeMap::new(),
-            };
+            let family = DistributionBucketFamily::<T>::default();
 
             let family_id = Self::next_distribution_bucket_family_id();
 
@@ -1890,9 +1937,11 @@ decl_module! {
             // == MUTATION SAFE ==
             //
 
-            let bucket = DistributionBucket {
+            let bucket = DistributionBucket::<T> {
                 accepting_new_bags,
                 distributing: true,
+                pending_invitations: BTreeSet::new(),
+                operators: BTreeSet::new(),
             };
 
             let bucket_id = Self::next_distribution_bucket_id();
@@ -2098,6 +2147,43 @@ decl_module! {
             );
         }
 
+        /// Invite an operator. Must be missing.
+        #[weight = 10_000_000] // TODO: adjust weight
+        pub fn invite_distribution_bucket_operator(
+            origin,
+            distribution_bucket_family_id: T::DistributionBucketFamilyId,
+            distribution_bucket_id: T::DistributionBucketId,
+            worker_id: WorkerId<T>
+        ) {
+            T::ensure_distribution_working_group_leader_origin(origin)?;
+
+            let mut family =
+                Self::ensure_distribution_bucket_family_exists(&distribution_bucket_family_id)?;
+            let mut bucket = Self::ensure_distribution_bucket_exists(
+                &family,
+                &distribution_bucket_id
+            )?;
+
+            Self::ensure_distribution_provider_can_be_invited(&bucket, &worker_id)?;
+
+            //
+            // == MUTATION SAFE ==
+            //
+
+            bucket.pending_invitations.insert(worker_id);
+            family.distribution_buckets.insert(distribution_bucket_id, bucket);
+
+            <DistributionBucketFamilyById<T>>::insert(distribution_bucket_family_id, family);
+
+            Self::deposit_event(
+                RawEvent::DistributionBucketOperatorInvited(
+                    distribution_bucket_family_id,
+                    distribution_bucket_id,
+                    worker_id,
+                )
+            );
+        }
+
         // ===== Distribution Operator actions =====
 
     }
@@ -2925,7 +3011,7 @@ impl<T: Trait> Module<T> {
         Self::get_default_dynamic_bag_creation_policy(bag_type)
     }
 
-    // Verifies storage provider operator existence.
+    // Verifies storage operator existence.
     fn ensure_storage_provider_operator_exists(operator_id: &WorkerId<T>) -> DispatchResult {
         ensure!(
             T::ensure_storage_worker_exists(operator_id).is_ok(),
@@ -2939,7 +3025,7 @@ impl<T: Trait> Module<T> {
     // Returns the DistributionBucketFamily object or error.
     fn ensure_distribution_bucket_family_exists(
         family_id: &T::DistributionBucketFamilyId,
-    ) -> Result<DistributionBucketFamily<T::DistributionBucketId>, Error<T>> {
+    ) -> Result<DistributionBucketFamily<T>, Error<T>> {
         ensure!(
             <DistributionBucketFamilyById<T>>::contains_key(family_id),
             Error::<T>::DistributionBucketFamilyDoesntExist
@@ -2951,9 +3037,9 @@ impl<T: Trait> Module<T> {
     // Ensures the existence of the distribution bucket.
     // Returns the DistributionBucket object or error.
     fn ensure_distribution_bucket_exists(
-        family: &DistributionBucketFamily<T::DistributionBucketId>,
+        family: &DistributionBucketFamily<T>,
         distribution_bucket_id: &T::DistributionBucketId,
-    ) -> Result<DistributionBucket, Error<T>> {
+    ) -> Result<DistributionBucket<T>, Error<T>> {
         family
             .distribution_buckets
             .get(distribution_bucket_id)
@@ -3049,4 +3135,27 @@ impl<T: Trait> Module<T> {
             T::Randomness::random_seed()
         }
     }
+
+    // Verify parameters for the `invite_distribution_bucket_operator` extrinsic.
+    fn ensure_distribution_provider_can_be_invited(
+        bucket: &DistributionBucket<T>,
+        worker_id: &WorkerId<T>,
+    ) -> DispatchResult {
+        ensure!(
+            T::ensure_distribution_worker_exists(worker_id).is_ok(),
+            Error::<T>::DistributionProviderOperatorDoesntExist
+        );
+
+        ensure!(
+            !bucket.pending_invitations.contains(worker_id),
+            Error::<T>::DistributionProviderOperatorAlreadyInvited
+        );
+
+        ensure!(
+            !bucket.operators.contains(worker_id),
+            Error::<T>::DistributionProviderOperatorAlreadySet
+        );
+
+        Ok(())
+    }
 }

runtime-modules/storage/src/tests/fixtures.rs

@@ -1319,7 +1319,7 @@ impl CreateDistributionBucketFixture {
         if actual_result.is_ok() {
             assert_eq!(next_bucket_id + 1, Storage::next_distribution_bucket_id());
 
-            let family: DistributionBucketFamily<u64> =
+            let family: DistributionBucketFamily<Test> =
                 Storage::distribution_bucket_family_by_id(self.family_id);
 
             assert!(family.distribution_buckets.contains_key(&next_bucket_id));
@@ -1628,3 +1628,63 @@ impl UpdateFamiliesInDynamicBagCreationPolicyFixture {
         }
     }
 }
+
+pub struct InviteDistributionBucketOperatorFixture {
+    origin: RawOrigin<u64>,
+    operator_worker_id: u64,
+    family_id: u64,
+    bucket_id: u64,
+}
+
+impl InviteDistributionBucketOperatorFixture {
+    pub fn default() -> Self {
+        Self {
+            origin: RawOrigin::Signed(DEFAULT_ACCOUNT_ID),
+            operator_worker_id: DEFAULT_WORKER_ID,
+            bucket_id: Default::default(),
+            family_id: Default::default(),
+        }
+    }
+
+    pub fn with_origin(self, origin: RawOrigin<u64>) -> Self {
+        Self { origin, ..self }
+    }
+
+    pub fn with_operator_worker_id(self, operator_worker_id: u64) -> Self {
+        Self {
+            operator_worker_id,
+            ..self
+        }
+    }
+
+    pub fn with_bucket_id(self, bucket_id: u64) -> Self {
+        Self { bucket_id, ..self }
+    }
+
+    pub fn with_family_id(self, family_id: u64) -> Self {
+        Self { family_id, ..self }
+    }
+
+    pub fn call_and_assert(&self, expected_result: DispatchResult) {
+        let actual_result = Storage::invite_distribution_bucket_operator(
+            self.origin.clone().into(),
+            self.family_id,
+            self.bucket_id,
+            self.operator_worker_id,
+        );
+
+        assert_eq!(actual_result, expected_result);
+
+        if actual_result.is_ok() {
+            let new_family = Storage::distribution_bucket_family_by_id(self.family_id);
+            let new_bucket = new_family
+                .distribution_buckets
+                .get(&self.bucket_id)
+                .unwrap();
+
+            assert!(new_bucket
+                .pending_invitations
+                .contains(&self.operator_worker_id),);
+        }
+    }
+}

runtime-modules/storage/src/tests/mocks.rs

@@ -73,6 +73,8 @@ pub const DEFAULT_DISTRIBUTION_PROVIDER_ACCOUNT_ID: u64 = 100003;
 pub const DISTRIBUTION_WG_LEADER_ACCOUNT_ID: u64 = 100004;
 pub const DEFAULT_STORAGE_PROVIDER_ID: u64 = 10;
 pub const ANOTHER_STORAGE_PROVIDER_ID: u64 = 11;
+pub const DEFAULT_DISTRIBUTION_PROVIDER_ID: u64 = 12;
+pub const ANOTHER_DISTRIBUTION_PROVIDER_ID: u64 = 13;
 
 impl crate::Trait for Test {
     type Event = TestEvent;
@@ -80,6 +82,7 @@ impl crate::Trait for Test {
     type StorageBucketId = u64;
     type DistributionBucketId = u64;
     type DistributionBucketFamilyId = u64;
+    type DistributionBucketOperatorId = u64;
     type ChannelId = u64;
     type MaxStorageBucketNumber = MaxStorageBucketNumber;
     type MaxNumberOfDataObjectsPerBag = MaxNumberOfDataObjectsPerBag;
@@ -148,6 +151,19 @@ impl crate::Trait for Test {
             Ok(())
         }
     }
+
+    fn ensure_distribution_worker_exists(worker_id: &u64) -> DispatchResult {
+        let allowed_providers = vec![
+            DEFAULT_DISTRIBUTION_PROVIDER_ID,
+            ANOTHER_DISTRIBUTION_PROVIDER_ID,
+        ];
+
+        if !allowed_providers.contains(worker_id) {
+            Err(DispatchError::Other("Invalid worker"))
+        } else {
+            Ok(())
+        }
+    }
 }
 
 pub const DEFAULT_MEMBER_ID: u64 = 100;

runtime-modules/storage/src/tests/mod.rs

@@ -25,8 +25,9 @@ use mocks::{
     DefaultChannelDynamicBagNumberOfStorageBuckets, DefaultMemberDynamicBagNumberOfStorageBuckets,
     InitialStorageBucketsNumberForDynamicBag, MaxDistributionBucketFamilyNumber,
     MaxDistributionBucketNumberPerFamily, MaxNumberOfDataObjectsPerBag, MaxRandomIterationNumber,
-    MaxStorageBucketNumber, Storage, Test, ANOTHER_STORAGE_PROVIDER_ID, DEFAULT_MEMBER_ACCOUNT_ID,
-    DEFAULT_MEMBER_ID, DEFAULT_STORAGE_PROVIDER_ACCOUNT_ID, DEFAULT_STORAGE_PROVIDER_ID,
+    MaxStorageBucketNumber, Storage, Test, ANOTHER_STORAGE_PROVIDER_ID,
+    DEFAULT_DISTRIBUTION_PROVIDER_ID, DEFAULT_MEMBER_ACCOUNT_ID, DEFAULT_MEMBER_ID,
+    DEFAULT_STORAGE_PROVIDER_ACCOUNT_ID, DEFAULT_STORAGE_PROVIDER_ID,
     DISTRIBUTION_WG_LEADER_ACCOUNT_ID, STORAGE_WG_LEADER_ACCOUNT_ID,
 };
 
@@ -3496,7 +3497,7 @@ fn create_distribution_bucket_family_succeeded() {
 
         let bucket_family = Storage::distribution_bucket_family_by_id(family_id);
 
-        assert_eq!(bucket_family, DistributionBucketFamily::default());
+        assert_eq!(bucket_family, DistributionBucketFamily::<Test>::default());
 
         EventFixture::assert_last_crate_event(RawEvent::DistributionBucketFamilyCreated(family_id));
     });
@@ -4205,3 +4206,124 @@ fn distribution_bucket_family_pick_during_dynamic_bag_creation_succeeded() {
         assert_eq!(total_ids1, total_ids2); // picked IDS are from total ID set.
     });
 }
+
+#[test]
+fn invite_distribution_bucket_operator_succeeded() {
+    build_test_externalities().execute_with(|| {
+        let starting_block = 1;
+        run_to_block(starting_block);
+
+        let provider_id = DEFAULT_DISTRIBUTION_PROVIDER_ID;
+
+        let family_id = CreateDistributionBucketFamilyFixture::default()
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .call_and_assert(Ok(()))
+            .unwrap();
+
+        let bucket_id = CreateDistributionBucketFixture::default()
+            .with_family_id(family_id)
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .call_and_assert(Ok(()))
+            .unwrap();
+
+        InviteDistributionBucketOperatorFixture::default()
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .with_bucket_id(bucket_id)
+            .with_family_id(family_id)
+            .with_operator_worker_id(provider_id)
+            .call_and_assert(Ok(()));
+
+        EventFixture::assert_last_crate_event(RawEvent::DistributionBucketOperatorInvited(
+            family_id,
+            bucket_id,
+            provider_id,
+        ));
+    });
+}
+
+#[test]
+fn invite_distribution_bucket_operator_fails_with_non_leader_origin() {
+    build_test_externalities().execute_with(|| {
+        let non_leader_id = 1;
+
+        InviteDistributionBucketOperatorFixture::default()
+            .with_origin(RawOrigin::Signed(non_leader_id))
+            .call_and_assert(Err(DispatchError::BadOrigin));
+    });
+}
+
+#[test]
+fn invite_distribution_bucket_operator_fails_with_non_existing_storage_bucket() {
+    build_test_externalities().execute_with(|| {
+        let family_id = CreateDistributionBucketFamilyFixture::default()
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .call_and_assert(Ok(()))
+            .unwrap();
+
+        InviteDistributionBucketOperatorFixture::default()
+            .with_family_id(family_id)
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .call_and_assert(Err(Error::<Test>::DistributionBucketDoesntExist.into()));
+    });
+}
+
+#[test]
+fn invite_distribution_bucket_operator_fails_with_non_missing_invitation() {
+    build_test_externalities().execute_with(|| {
+        let invited_worker_id = DEFAULT_DISTRIBUTION_PROVIDER_ID;
+
+        let family_id = CreateDistributionBucketFamilyFixture::default()
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .call_and_assert(Ok(()))
+            .unwrap();
+
+        let bucket_id = CreateDistributionBucketFixture::default()
+            .with_family_id(family_id)
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .call_and_assert(Ok(()))
+            .unwrap();
+
+        InviteDistributionBucketOperatorFixture::default()
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .with_bucket_id(bucket_id)
+            .with_family_id(family_id)
+            .with_operator_worker_id(invited_worker_id)
+            .call_and_assert(Ok(()));
+
+        InviteDistributionBucketOperatorFixture::default()
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .with_bucket_id(bucket_id)
+            .with_family_id(family_id)
+            .with_operator_worker_id(invited_worker_id)
+            .call_and_assert(Err(
+                Error::<Test>::DistributionProviderOperatorAlreadyInvited.into(),
+            ));
+    });
+}
+
+#[test]
+fn invite_distribution_bucket_operator_fails_with_invalid_storage_provider_id() {
+    build_test_externalities().execute_with(|| {
+        let invalid_provider_id = 155;
+
+        let family_id = CreateDistributionBucketFamilyFixture::default()
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .call_and_assert(Ok(()))
+            .unwrap();
+
+        let bucket_id = CreateDistributionBucketFixture::default()
+            .with_family_id(family_id)
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .call_and_assert(Ok(()))
+            .unwrap();
+
+        InviteDistributionBucketOperatorFixture::default()
+            .with_origin(RawOrigin::Signed(DISTRIBUTION_WG_LEADER_ACCOUNT_ID))
+            .with_bucket_id(bucket_id)
+            .with_family_id(family_id)
+            .with_operator_worker_id(invalid_provider_id)
+            .call_and_assert(Err(
+                Error::<Test>::DistributionProviderOperatorDoesntExist.into()
+            ));
+    });
+}

runtime/src/lib.rs

@@ -691,6 +691,7 @@ impl storage::Trait for Runtime {
     type MaxDistributionBucketFamilyNumber = MaxDistributionBucketFamilyNumber;
     type MaxDistributionBucketNumberPerFamily = MaxDistributionBucketNumberPerFamily;
     type DistributionBucketsPerBagValueConstraint = DistributionBucketsPerBagValueConstraint;
+    type DistributionBucketOperatorId = DistributionBucketOperatorId;
 
     fn ensure_storage_working_group_leader_origin(origin: Self::Origin) -> DispatchResult {
         StorageWorkingGroup::ensure_origin_is_active_leader(origin)
@@ -716,6 +717,12 @@ impl storage::Trait for Runtime {
     ) -> DispatchResult {
         DistributionWorkingGroup::ensure_worker_signed(origin, &worker_id).map(|_| ())
     }
+
+    fn ensure_distribution_worker_exists(worker_id: &ActorId) -> DispatchResult {
+        DistributionWorkingGroup::ensure_worker_exists(&worker_id)
+            .map(|_| ())
+            .map_err(|err| err.into())
+    }
 }
 
 /// Opaque types. These are used by the CLI to instantiate machinery that don't need to know

runtime/src/primitives.rs

@@ -80,6 +80,9 @@ pub type DistributionBucketFamilyId = u64;
 /// Represent a media channel.
 pub type ChannelId = u64;
 
+/// Represent relationships between distribution buckets and distribution working group workers.
+pub type DistributionBucketOperatorId = u64;
+
 /// App-specific crypto used for reporting equivocation/misbehavior in BABE and
 /// GRANDPA. Any rewards for misbehavior reporting will be paid out to this
 /// account.