feat: rearrange submissions storage map and apply more submission limits

Cargo.toml

@@ -144,6 +144,7 @@ pallet-balances = { version = "39.0.0", default-features = false }
 pallet-grandpa = { version = "38.0.0", default-features = false }
 pallet-sudo = { version = "38.0.0", default-features = false }
 pallet-multisig = { version = "38.0.0", default-features = false }
+pallet-migrations = { version = "8.0.0", default-features = false }
 pallet-timestamp = { version = "37.0.0", default-features = false }
 pallet-transaction-payment-rpc-runtime-api = { version = "38.0.0", default-features = false }
 scale-info = { version = "2.11.1", default-features = false }

pallets/indexing/Cargo.toml

@@ -23,11 +23,13 @@ scale-info = { features = [
 frame-benchmarking = { optional = true, workspace = true }
 frame-support.workspace = true
 frame-system.workspace = true
+itertools.workspace = true
 pallet-commitments.workspace = true
 pallet-permissions.workspace = true
 pallet-tables.workspace = true
 pallet-system-tables.workspace = true
 pallet-session.workspace = true
+pallet-migrations.workspace = true
 hex.workspace = true
 sxt-core.workspace = true
 sp-runtime = { workspace = true, default-features = false }
@@ -70,6 +72,7 @@ std = [
 	"pallet-balances/std",
 	"pallet-staking/std",
 	"pallet-system-tables/std",
+	"pallet-migrations/std",
 ]
 runtime-benchmarks = [
 	"frame-benchmarking/runtime-benchmarks",
@@ -78,9 +81,11 @@ runtime-benchmarks = [
 	"pallet-staking/runtime-benchmarks",
 	"pallet-system-tables/runtime-benchmarks",
 	"sp-staking/runtime-benchmarks",
+	"pallet-migrations/runtime-benchmarks",
 	"dep:proof-of-sql-commitment-map",
 ]
 try-runtime = [
 	"frame-support/try-runtime",
 	"frame-system/try-runtime",
+	"pallet-migrations/try-runtime",
 ]

pallets/indexing/src/benchmarking.rs

@@ -3,6 +3,7 @@ use alloc::vec;
 
 use frame_benchmarking::v2::*;
 use frame_system::RawOrigin;
+use sp_core::Hasher;
 
 use super::*;
 #[cfg(test)]
@@ -13,12 +14,15 @@ use crate::Pallet as Indexing;
 #[allow(clippy::multiple_bound_locations)]
 #[instance_benchmarks(where I: NativeApi)]
 mod benchmarks {
+    use frame_support::migrations::SteppedMigration;
+    use frame_support::weights::WeightMeter;
     use native_api::NativeApi;
     use pallet_tables::{CommitmentCreationCmd, UpdateTable};
     use proof_of_sql_commitment_map::CommitmentSchemeFlags;
     use sxt_core::permissions::{IndexingPalletPermission, PermissionLevel, PermissionList};
     use sxt_core::tables::{
         InsertQuorumSize,
+        QuorumScope,
         Source,
         TableIdentifier,
         TableName,
@@ -162,6 +166,66 @@ mod benchmarks {
         assert!(Indexing::<T, I>::final_data(batch_id).is_some());
     }
 
+    #[benchmark]
+    fn migration_v0_v1_step() {
+        let submitters_by_scope = (0..MAX_SUBMITTERS * 2)
+            .map(|submitter_num| {
+                let scope = if submitter_num % 2 == 0 {
+                    QuorumScope::Public
+                } else {
+                    QuorumScope::Privileged
+                };
+
+                (account("submitter", submitter_num, 0), scope)
+            })
+            .fold(
+                SubmittersByScope::default(),
+                |submitters_by_scope, (submitter, scope)| {
+                    submitters_by_scope
+                        .with_submitter(submitter, &scope)
+                        .unwrap()
+                },
+            );
+        let hash = <<T as frame_system::Config>::Hashing as Hasher>::hash(&[]);
+        crate::migrations::v1::v0::Submissions::<T, I>::insert(
+            BatchId::default(),
+            hash,
+            submitters_by_scope,
+        );
+        let mut meter = WeightMeter::new();
+
+        #[block]
+        {
+            crate::migrations::v1::LazyMigrationV1::<T, weights::SubstrateWeight<T>, I>::step(
+                None, &mut meter,
+            )
+            .unwrap();
+        }
+
+        // Check that the new storage is decodable:
+        (0..MAX_SUBMITTERS * 2).for_each(|submitter_num| {
+            let scope = if submitter_num % 2 == 0 {
+                QuorumScope::Public
+            } else {
+                QuorumScope::Privileged
+            };
+            assert_eq!(
+                crate::SubmissionsV1::<T, I>::get((
+                    BatchId::default(),
+                    scope,
+                    account::<T::AccountId>("submitter", submitter_num, 0)
+                ))
+                .unwrap(),
+                hash
+            );
+        });
+        // uses twice the weight once for migration and then for checking if there is another key.
+        assert_eq!(
+            meter.consumed(),
+            weights::SubstrateWeight::<T>::migration_v0_v1_step() * 2
+        );
+    }
+
     impl_benchmark_test_suite!(
         PalletWithApi,
         crate::mock::new_test_ext(),

pallets/indexing/src/lib.rs

@@ -31,9 +31,12 @@ mod error_conversions;
 /// Native wrapper around the indexing pallet.
 pub mod native_pallet;
 
+pub mod migrations;
+
 #[allow(clippy::manual_inspect)]
 #[frame_support::pallet]
 pub mod pallet {
+    use alloc::collections::{BTreeMap, BTreeSet};
     use alloc::string::String;
     use alloc::vec::Vec;
 
@@ -44,6 +47,7 @@ pub mod pallet {
     use frame_support::{Blake2_128, Blake2_128Concat};
     use frame_system::pallet_prelude::*;
     use hex::FromHex;
+    use itertools::{Either, Itertools};
     use native_api::NativeApi;
     use on_chain_table::OnChainTable;
     use sp_core::{H256, U256};
@@ -80,19 +84,17 @@ pub mod pallet {
         type WeightInfo: WeightInfo;
     }
 
-    /// Double Map of Submissions using the batch-id as the first key and the submitter's
-    /// public key as the second key to hold the hash of the submitted data.
-    /// Each submission for a given batch id will have an entry here
+    /// Hashed data submissions that have yet to reach quorum.
     #[pallet::storage]
     #[pallet::getter(fn submissions)]
-    pub type Submissions<T: Config<I>, I: 'static = ()> = StorageDoubleMap<
+    pub type SubmissionsV1<T: Config<I>, I: 'static = ()> = StorageNMap<
         _,
-        Blake2_128Concat,
-        BatchId,
-        Blake2_128Concat,
+        (
+            NMapKey<Blake2_128Concat, BatchId>,
+            NMapKey<Blake2_128Concat, QuorumScope>,
+            NMapKey<Blake2_128Concat, T::AccountId>,
+        ),
         <T as frame_system::Config>::Hash,
-        SubmittersByScope<T::AccountId>,
-        ValueQuery, // Allows us to receive a default instead of None
     >;
 
     #[pallet::storage]
@@ -170,8 +172,6 @@ pub mod pallet {
         LateBatch,
         /// Invalid Table identifier was supplied
         InvalidTable,
-        /// This user has already submitted data for this batch id
-        AlreadySubmitted,
         /// The table could not be deserialized using a Stream Reader
         NativeDeserializationError,
         /// There was no record batch contained in the data
@@ -190,6 +190,8 @@ pub mod pallet {
         NativeRecordBatchDuplicateIdentifiers,
         /// Error serializing the OnChainTable
         NativeSerializationError,
+        /// Maximum submissions already reached for this batch id
+        MaxSubmittersReached,
         /// Error deserializing the table as an OnChainTable
         TableDeserializationError,
         /// Error deserializing the table as an OnChainTable
@@ -327,61 +329,58 @@ pub mod pallet {
         T: Config<I>,
         I: NativeApi,
     {
-        // We don't need to save the full data. We just need a count associated with each submission
-        let match_submissions = Submissions::<T, I>::get(&batch_id, data_hash);
-
-        // Check if this user has already submitted this data
-        let current_num_matching_submissions = match_submissions.len_of_scope(quorum_scope);
-
-        let new_match_submissions = match_submissions
-            .with_submitter(who.clone(), quorum_scope)
-            // Just return the unchanged submissions if maximum is exceeded.
-            .unwrap_or_else(|(submitters, _)| submitters);
-
-        if new_match_submissions.len_of_scope(quorum_scope) == current_num_matching_submissions {
-            Err(Error::<T, I>::AlreadySubmitted)?
+        let submission_map_with_this =
+            SubmissionsV1::<T, I>::iter_prefix((&batch_id, quorum_scope))
+                .take(MAX_SUBMITTERS as usize)
+                .chain(core::iter::once((who.clone(), data_hash)))
+                .collect::<BTreeMap<_, _>>();
+
+        if submission_map_with_this.len() > MAX_SUBMITTERS as usize {
+            Err(Error::MaxSubmittersReached::<T, I>)?;
         }
-
-        Submissions::<T, I>::insert(&batch_id, data_hash, &new_match_submissions);
+        SubmissionsV1::<T, I>::insert((&batch_id, quorum_scope, &who), data_hash);
 
         let submission = DataSubmission {
             table: table.clone(),
             batch_id: batch_id.clone(),
             data_hash,
             quorum_scope: *quorum_scope,
         };
-
         // Emit an event noting who submitted what
         Pallet::<T, I>::deposit_event(Event::DataSubmitted { who, submission });
 
-        match table_insert_quorum.of_scope(quorum_scope) {
-            Some(quorum_size)
-                if new_match_submissions.len_of_scope(quorum_scope) as u8 > *quorum_size =>
-            {
-                // Iterate over the submitters who submitted differing data and collect
-                // their account ids
-                let dissenters = Submissions::<T, I>::iter_prefix(&batch_id)
-                    .filter(|(hash, _)| hash != &data_hash)
-                    .flat_map(|(_, submitters)| submitters.into_iter_scope(quorum_scope))
-                    // de-dup collection
-                    .collect::<alloc::collections::BTreeSet<_>>()
-                    .into_iter()
-                    // resulting set should contain up to MAX_SUBMITTERS items *after* de-dup
-                    .take(MAX_SUBMITTERS as usize)
-                    .collect::<alloc::collections::BTreeSet<_>>()
-                    .try_into()
-                    .expect("source Vec is constructed to not exceed maximum submitter list size");
+        let (agreements_unbounded, dissents_unbounded): (BTreeSet<_>, BTreeSet<_>) =
+            submission_map_with_this
+                .into_iter()
+                .partition_map(|(account_id, hash)| {
+                    if hash == data_hash {
+                        Either::Left(account_id)
+                    } else {
+                        Either::Right(account_id)
+                    }
+                });
 
+        match table_insert_quorum.of_scope(quorum_scope) {
+            Some(quorum_size) if agreements_unbounded.len() as u8 > *quorum_size => {
                 let block_number = <frame_system::Pallet<T>>::block_number();
 
+                const ALREADY_VALIDATED_SUBMITTER_COUNT: &str =
+                    "we've already validated that the submitter count is within bounds";
+                let agreements = agreements_unbounded
+                    .try_into()
+                    .expect(ALREADY_VALIDATED_SUBMITTER_COUNT);
+                let dissents = dissents_unbounded
+                    .try_into()
+                    .expect(ALREADY_VALIDATED_SUBMITTER_COUNT);
+
                 // Decide on the quorum
                 let data_quorum = DataQuorum {
                     table,
                     batch_id,
                     data_hash,
                     block_number: block_number.into(),
-                    agreements: new_match_submissions.of_scope(quorum_scope).clone(),
-                    dissents: dissenters,
+                    agreements,
+                    dissents,
                     quorum_scope: *quorum_scope,
                 };
 
@@ -405,8 +404,11 @@ pub mod pallet {
         T: Config<I>,
         I: NativeApi,
     {
-        Submissions::<T, I>::iter_key_prefix(&quorum.batch_id)
-            .for_each(|key| Submissions::<T, I>::remove(&quorum.batch_id, key));
+        SubmissionsV1::<T, I>::clear_prefix(
+            (&quorum.batch_id,),
+            MAX_SUBMITTERS * QuorumScope::VARIANT_COUNT as u32,
+            None,
+        );
 
         FinalData::<T, I>::insert(&quorum.batch_id, &quorum);
 

pallets/indexing/src/migrations/mod.rs

@@ -0,0 +1,11 @@
+//! Storage migrations for the indexing pallet
+
+pub mod v1;
+
+/// A unique identifier across all pallets.
+///
+/// This constant represents a unique identifier for the migrations of this pallet.
+/// It helps differentiate migrations for this pallet from those of others. Note that we don't
+/// directly pull the crate name from the environment, since that would change if the crate were
+/// ever to be renamed and could cause historic migrations to run again.
+pub const PALLET_MIGRATIONS_ID: &[u8; 26] = b"pallet-indexing-migrations";