feat: migrate storage to populate batch queue from submissions

Recently, the indexing pallet imposed a bound on batches finding quorum
with the help of a new BatchQueue in storage. This adds a storage
migration to populate the initial contents of this BatchQueue, by
analyzing the submissions storage.

This implementation is simple but dumb. There's no way for us to know
what order the batches were initially submitted in from current storage,
so this just inserts them to the batch queue in their iteration order
(lexicographic ordering of the hash of the batch ids). Furthermore, this
does not do any pruning, so ideally this the batch queue capacity should
exceed the number of batches finding quorum in all live networks so that
they are pruned more gradually.

Author: tlovell-sxt

pallets/indexing/src/benchmarking.rs

@@ -226,6 +226,46 @@ mod benchmarks {
         );
     }
 
+    #[benchmark]
+    fn migration_v1_v2_step() {
+        let hash = <<T as frame_system::Config>::Hashing as Hasher>::hash(&[]);
+        let batch_id = BatchId::default();
+        (0..MAX_SUBMITTERS * 2)
+            .map(|submitter_num| {
+                let scope = if submitter_num % 2 == 0 {
+                    QuorumScope::Public
+                } else {
+                    QuorumScope::Privileged
+                };
+
+                (
+                    account::<T::AccountId>("submitter", submitter_num, 0),
+                    scope,
+                )
+            })
+            .for_each(|(account_id, scope)| {
+                crate::SubmissionsV1::<T, I>::insert((&batch_id, scope, account_id), hash);
+            });
+
+        let mut meter = WeightMeter::new();
+
+        #[block]
+        {
+            crate::migrations::v2::LazyMigrationV2::<T, weights::SubstrateWeight<T>, I>::step(
+                None, &mut meter,
+            )
+            .unwrap();
+        }
+
+        // Check that the new storage is decodable:
+        assert_eq!(crate::BatchQueue::<T, I>::get(0).unwrap(), batch_id);
+        // uses twice the weight once for migration and then for checking if there is another key.
+        assert_eq!(
+            meter.consumed(),
+            weights::SubstrateWeight::<T>::migration_v1_v2_step() * 2
+        );
+    }
+
     impl_benchmark_test_suite!(
         PalletWithApi,
         crate::mock::new_test_ext(),

pallets/indexing/src/lib.rs

@@ -586,7 +586,7 @@ pub mod pallet {
             .next()
             .is_some()
         {
-            return vec![];
+            return Vec::new();
         }
 
         let batches_to_prune = push_batch_to_queue_and_prune::<T, I>(batch_id);

pallets/indexing/src/migrations/mod.rs

@@ -2,6 +2,8 @@
 
 pub mod v1;
 
+pub mod v2;
+
 /// A unique identifier across all pallets.
 ///
 /// This constant represents a unique identifier for the migrations of this pallet.

pallets/indexing/src/migrations/v2/mod.rs

@@ -0,0 +1,92 @@
+//! Migration from pallet-indexing storage v1 to v2
+//!
+//! This migration handles changes to the `Submitters`, which went from being a mapping of data
+//! hashes to submitters to the reverse.
+
+use frame_support::migrations::{MigrationId, SteppedMigration, SteppedMigrationError};
+use frame_support::pallet_prelude::PhantomData;
+use frame_support::weights::WeightMeter;
+use sxt_core::indexing::BatchId;
+use sxt_core::tables::QuorumScope;
+
+use super::PALLET_MIGRATIONS_ID;
+use crate::pallet::{Config, SubmissionsV1};
+
+mod tests;
+
+/// Populates the `BatchQueue` from existing `Submissions`.
+///
+/// As we cannot know the order that the batches were submitted from current storage information,
+/// the batches will be represented in the queue in their `SubmissionsV1` iteration order
+/// (lexicographic).
+///
+/// This will not do any pruning, even if current submissions exceed the `BATCH_QUEUE_CAPACITY`.
+/// So, it is recommended to start using the batch queue with a capacity higher than any live
+/// network's current submission count, to avoid a bunch of batches getting pruned on the very next
+/// submission.
+pub struct LazyMigrationV2<T: Config<I>, W: crate::weights::WeightInfo, I: 'static = ()>(
+    PhantomData<(T, W, I)>,
+);
+
+impl<T: Config<I>, W: crate::weights::WeightInfo, I: 'static> SteppedMigration
+    for LazyMigrationV2<T, W, I>
+{
+    type Cursor = (BatchId, QuorumScope, T::AccountId);
+    type Identifier = MigrationId<26>;
+
+    fn id() -> Self::Identifier {
+        MigrationId {
+            pallet_id: *PALLET_MIGRATIONS_ID,
+            version_from: 1,
+            version_to: 2,
+        }
+    }
+
+    fn step(
+        mut cursor: Option<Self::Cursor>,
+        meter: &mut WeightMeter,
+    ) -> Result<Option<Self::Cursor>, SteppedMigrationError> {
+        let required = W::migration_v1_v2_step();
+
+        if meter.remaining().any_lt(required) {
+            return Err(SteppedMigrationError::InsufficientWeight { required });
+        }
+
+        // We loop here to do as much progress as possible per step.
+        loop {
+            if meter.try_consume(required).is_err() {
+                break;
+            }
+
+            let mut iter = if let Some(key) = &cursor {
+                // If a cursor is provided, start iterating from the stored value
+                SubmissionsV1::<T, I>::iter_keys_from(SubmissionsV1::<T, I>::hashed_key_for(key))
+            } else {
+                // If no cursor is provided, start iterating from the beginning.
+                SubmissionsV1::<T, I>::iter_keys()
+            };
+
+            // If there's a next item in the iterator, perform the migration.
+            if let Some((batch_id, scope, account)) = iter.next() {
+                let batch_index = crate::BatchQueue::<T, I>::count();
+                crate::BatchQueue::<T, I>::insert(batch_index as u64, batch_id.clone());
+
+                // update cursor always
+                cursor = Some((batch_id.clone(), scope, account));
+
+                // update cursor to the last key w/ this batch id, skipping them in the migration
+                for key in iter {
+                    if key.0 == batch_id {
+                        cursor = Some(key);
+                    } else {
+                        break;
+                    }
+                }
+            } else {
+                cursor = None; // Signal that the migration is complete (no more items to process).
+                break;
+            }
+        }
+        Ok(cursor)
+    }
+}

pallets/indexing/src/migrations/v2/tests.rs

@@ -0,0 +1,131 @@
+#![cfg(all(test, not(feature = "runtime-benchmarks")))]
+
+use codec::Decode;
+use frame_support::traits::OnRuntimeUpgrade;
+use frame_system::ensure_signed;
+use native_api::Api;
+use pallet_migrations::WeightInfo as _;
+use sp_core::Hasher;
+use sxt_core::indexing::{BatchId, SubmittersByScope};
+use sxt_core::tables::QuorumScope;
+
+use crate::migrations::v1;
+use crate::mock::{
+    new_test_ext,
+    run_to_block,
+    AllPalletsWithSystem,
+    MigratorServiceWeight,
+    RuntimeOrigin,
+    System,
+    Test,
+};
+use crate::weights::WeightInfo as _;
+
+fn submissions_from_seed(
+    seed: u64,
+) -> (
+    BatchId,
+    <Test as frame_system::Config>::Hash,
+    SubmittersByScope<<Test as frame_system::Config>::AccountId>,
+) {
+    let batch_id = BatchId::try_from(seed.to_le_bytes().to_vec()).unwrap();
+    let data_hash = <<Test as frame_system::Config>::Hashing as Hasher>::hash(&seed.to_le_bytes());
+
+    let num_privileged_submissions = seed % 32;
+
+    // this math is chosen to
+    // - have some variety with num_privileged_submissions
+    // - always have at least 1 public submission (simplifies test logic)
+    let num_public_submissions = ((seed * 2) % 31) + 1;
+
+    let privileged_submitters_by_scope = (0..num_privileged_submissions)
+        .map(|submitter_seed_index| {
+            let submitter_index = seed * 32 + submitter_seed_index;
+            ensure_signed(RuntimeOrigin::signed(submitter_index)).unwrap()
+        })
+        .fold(SubmittersByScope::default(), |acc, submitter| {
+            acc.with_submitter(submitter, &QuorumScope::Privileged)
+                .unwrap()
+        });
+    let submitters_by_scope = (0..num_public_submissions)
+        .map(|submitter_seed_index| {
+            let submitter_index = seed * 32 + submitter_seed_index;
+            ensure_signed(RuntimeOrigin::signed(submitter_index)).unwrap()
+        })
+        .fold(privileged_submitters_by_scope, |acc, submitter| {
+            acc.with_submitter(submitter, &QuorumScope::Public).unwrap()
+        });
+
+    (batch_id, data_hash, submitters_by_scope)
+}
+
+#[test]
+fn lazy_migration_works() {
+    new_test_ext().execute_with(|| {
+        (0..64u64)
+            .map(submissions_from_seed)
+            .for_each(|(batch_id, data_hash, submitters)| {
+                submitters
+                    .iter_scope(&QuorumScope::Public)
+                    .for_each(|submitter| {
+                        crate::SubmissionsV1::<Test, Api>::insert(
+                            (&batch_id, QuorumScope::Public, &submitter),
+                            &data_hash,
+                        );
+                    });
+                submitters
+                    .iter_scope(&QuorumScope::Privileged)
+                    .for_each(|submitter| {
+                        crate::SubmissionsV1::<Test, Api>::insert(
+                            (&batch_id, QuorumScope::Privileged, &submitter),
+                            &data_hash,
+                        );
+                    });
+            });
+
+        // Give it enough weight do do exactly 8 iterations:
+        let limit = <Test as pallet_migrations::Config>::WeightInfo::progress_mbms_none()
+            + pallet_migrations::Pallet::<Test>::exec_migration_max_weight()
+            + crate::weights::SubstrateWeight::<Test>::migration_v1_v2_step() * 8;
+        MigratorServiceWeight::set(&limit);
+
+        System::set_block_number(1);
+        AllPalletsWithSystem::on_runtime_upgrade(); // onboard MBMs
+
+        // check migration progress across many blocks
+        let mut last_num_migrated = 0;
+        for block in 2..=10 {
+            run_to_block(block);
+
+            let num_migrated = (0..64)
+                .map(submissions_from_seed)
+                .map(|(seeded_batch_id, _, _)| {
+                    if crate::BatchQueue::<Test, Api>::iter()
+                        .any(|(_, batch_id)| seeded_batch_id == batch_id)
+                    {
+                        1usize
+                    } else {
+                        0usize
+                    }
+                })
+                .sum();
+
+            // part of the first block of migration is completing lazy migration v1 (noop), but
+            // this costs some weight so we cannot migrate 8 batches in the first block.
+            let expected_migrated_number = if block == 2 || block == 10 { 4 } else { 8 };
+
+            assert_eq!(num_migrated, last_num_migrated + expected_migrated_number);
+            last_num_migrated = num_migrated;
+        }
+
+        // Check that everything is migrated now
+        (0..64u64)
+            .map(submissions_from_seed)
+            .for_each(|(seeded_batch_id, _, _)| {
+                assert!(crate::BatchQueue::<Test, Api>::iter()
+                    .any(|(_, batch_id)| seeded_batch_id == batch_id))
+            });
+
+        assert_eq!(crate::BatchQueue::<Test, Api>::count(), 64);
+    });
+}

pallets/indexing/src/mock.rs

@@ -54,6 +54,7 @@ impl pallet_migrations::Config for Test {
     #[cfg(not(feature = "runtime-benchmarks"))]
     type Migrations = (
         crate::migrations::v1::LazyMigrationV1<Test, crate::weights::SubstrateWeight<Test>, Api>,
+        crate::migrations::v2::LazyMigrationV2<Test, crate::weights::SubstrateWeight<Test>, Api>,
     );
     #[cfg(feature = "runtime-benchmarks")]
     type Migrations = pallet_migrations::mock_helpers::MockedMigrations;

pallets/indexing/src/weights.rs

@@ -59,6 +59,13 @@ pub trait WeightInfo {
     /// Storage: `Indexing::SubmissionsV1` (r:0 w:64)
     /// Proof: `Indexing::SubmissionsV1` (`max_values`: None, `max_size`: Some(150), added: 2625, mode: `MaxEncodedLen`)
     fn migration_v0_v1_step() -> Weight;
+    /// Storage: `Indexing::SubmissionsV1` (r:65 w:0)
+    /// Proof: `Indexing::SubmissionsV1` (`max_values`: None, `max_size`: Some(150), added: 2625, mode: `MaxEncodedLen`)
+    /// Storage: `Indexing::CounterForBatchQueue` (r:1 w:1)
+    /// Proof: `Indexing::CounterForBatchQueue` (`max_values`: Some(1), `max_size`: Some(4), added: 499, mode: `MaxEncodedLen`)
+    /// Storage: `Indexing::BatchQueue` (r:1 w:1)
+    /// Proof: `Indexing::BatchQueue` (`max_values`: None, `max_size`: Some(45), added: 2520, mode: `MaxEncodedLen`)
+    fn migration_v1_v2_step() -> Weight;
 }
 
 pub struct SubstrateWeight<T>(PhantomData<T>);
@@ -133,4 +140,20 @@ impl<T: frame_system::Config> WeightInfo for SubstrateWeight<T> {
             .saturating_add(T::DbWeight::get().reads(2))
             .saturating_add(T::DbWeight::get().writes(65))
     }
+    /// Storage: `Indexing::SubmissionsV1` (r:65 w:0)
+    /// Proof: `Indexing::SubmissionsV1` (`max_values`: None, `max_size`: Some(150), added: 2625, mode: `MaxEncodedLen`)
+    /// Storage: `Indexing::CounterForBatchQueue` (r:1 w:1)
+    /// Proof: `Indexing::CounterForBatchQueue` (`max_values`: Some(1), `max_size`: Some(4), added: 499, mode: `MaxEncodedLen`)
+    /// Storage: `Indexing::BatchQueue` (r:1 w:1)
+    /// Proof: `Indexing::BatchQueue` (`max_values`: None, `max_size`: Some(45), added: 2520, mode: `MaxEncodedLen`)
+    fn migration_v1_v2_step() -> Weight {
+        // Proof Size summary in bytes:
+        //  Measured:  `6277`
+        //  Estimated: `171615`
+        // Minimum execution time: 133_010_000 picoseconds.
+        Weight::from_parts(135_916_000, 0)
+            .saturating_add(Weight::from_parts(0, 171615))
+            .saturating_add(T::DbWeight::get().reads(67))
+            .saturating_add(T::DbWeight::get().writes(2))
+    }
 }

runtime/src/lib.rs

@@ -341,6 +341,11 @@ impl pallet_migrations::Config for Runtime {
             pallet_indexing::SubstrateWeight<Runtime>,
             native_api::Api,
         >,
+        pallet_indexing::migrations::v2::LazyMigrationV2<
+            Runtime,
+            pallet_indexing::SubstrateWeight<Runtime>,
+            native_api::Api,
+        >,
     );
     // Benchmarks need mocked migrations to guarantee that they succeed.
     #[cfg(feature = "runtime-benchmarks")]