Add chunked submission strategy for multi-block solutions

The current implementation submits all pages concurrently, which can be
fastest if all submissions succeed quickly and the network/node can
handle the load efficiently. However, if an early transaction fails
(e.g., nonce N+1), all subsequent transactions will fail due to the
nonce gap, resulting in wasted submission attempts and potentially
high load on the network/node due to retransmissions.

This commit adds support for a `chunked submission strategy` where the
size of the chunk can be specified in the `experimental monitor
command`. If the chunk size is not specified or zero, we use the
fully concurrent approach. Otherwise, we submit a chunk of N pages,
wait for all of them to be successfully included in a block, and then
move to the next chunk.

This offers a good balance between speed and robustness/load. Note
that if the chunk size equals 1, we end up with a fully sequential
solution (submit 1 page - wait - submit next page, etc.).

Testing against a simple zombienet setup with an 8-page solution:

| Submission Strategy | Time (seconds) |
|---------------------|----------------|
| Fully concurrent    | ~77s           |
| Chunked (size=4)    | ~105s          |

Author: sigurpol

README.md

@@ -7,7 +7,7 @@ This is a re-write of the [staking miner](https://github.com/paritytech/polkadot
 The miner can work with two kinds of nodes:
 
 - **multi-phase (legacy)** , targeting the multi-phase, single paged election provider pallet
-- **multi-block (experimental) **, targeting the multi-phase, multi-block election provider pallet.
+- **multi-block (experimental)**, targeting the multi-phase, multi-block election provider pallet.
 
 The legacy path is intended to be temporary until all chains are migrated to multi-page staking (`asset-hub`).
 Once that occurs, we can remove the legacy components and rename the experimental monitor command to `monitor`.
@@ -104,6 +104,18 @@ This command is similar to the stable `monitor command` but targets the new pall
 polkadot-staking-miner --uri ws://127.0.0.1:9966 experimental-monitor-multi-block --seed-or-path //Alice
 ```
 
+The `--chunk-size` option controls how many solution pages are submitted concurrently. When set to 0 (default), all pages are submitted at once. When set to a positive number, pages are submitted in chunks of that size, waiting for each chunk to be included in a block before submitting the next chunk. This can help manage network load and improve reliability on congested networks or if the pages per solution increases.
+
+```bash
+polkadot-staking-miner --uri ws://127.0.0.1:9966 experimental-monitor-multi-block --seed-or-path //Alice --chunk-size 4
+```
+
+The `--do-reduce` option (off by default) enables solution reduction, to prevent further trimming, making submission more efficient.
+
+```bash
+polkadot-staking-miner --uri ws://127.0.0.1:9966 experimental-monitor-multi-block --seed-or-path //Alice --do-reduce
+```
+
 ### Prepare your SEED
 
 While you could pass your seed directly to the cli or Docker, this is highly **NOT** recommended. Instead, you should use an ENV variable.

src/commands/multi_block/monitor.rs

@@ -31,7 +31,7 @@ pub async fn monitor_cmd<T>(
 ) -> Result<(), Error>
 where
     T: MinerConfig<AccountId = AccountId> + Send + Sync + 'static,
-    T::Solution: Send,
+    T::Solution: Send + Sync + 'static,
     T::Pages: Send + Sync + 'static,
     T::TargetSnapshotPerBlock: Send + Sync + 'static,
     T::VoterSnapshotPerBlock: Send + Sync + 'static,
@@ -49,6 +49,7 @@ where
             .fetch(&runtime::storage().system().account(signer.account_id()))
             .await?
             .ok_or(Error::AccountDoesNotExists)?;
+
         prometheus::set_balance(account_info.data.free as f64);
 
         log::info!(
@@ -135,6 +136,7 @@ where
                 submit_lock,
                 config.submission_strategy,
                 config.do_reduce,
+                config.chunk_size,
             )
             .await
             {
@@ -165,14 +167,15 @@ async fn process_block<T>(
     submit_lock: Arc<Mutex<()>>,
     submission_strategy: SubmissionStrategy,
     do_reduce: bool,
+    chunk_size: usize,
 ) -> Result<(), Error>
 where
     T: MinerConfig<AccountId = AccountId> + Send + Sync + 'static,
-    T::Solution: Send,
-    T::Pages: Send,
+    T::Solution: Send + Sync + 'static,
+    T::Pages: Send + Sync + 'static,
     T::TargetSnapshotPerBlock: Send,
     T::VoterSnapshotPerBlock: Send,
-    T::MaxVotesPerVoter: Send,
+    T::MaxVotesPerVoter: Send + Sync + 'static,
 {
     let BlockDetails {
         storage,
@@ -268,7 +271,25 @@ where
                     missing_pages.push((page, solution));
                 }
 
-                dynamic::inner_submit_pages::<T>(&client, &signer, missing_pages, listen).await?;
+                // Use the appropriate submission method based on chunk_size
+                if chunk_size == 0 {
+                    dynamic::inner_submit_pages_concurrent::<T>(
+                        &client,
+                        &signer,
+                        missing_pages,
+                        listen,
+                    )
+                    .await?;
+                } else {
+                    dynamic::inner_submit_pages_chunked::<T>(
+                        &client,
+                        &signer,
+                        missing_pages,
+                        listen,
+                        chunk_size,
+                    )
+                    .await?;
+                }
                 return Ok(());
             }
 
@@ -301,7 +322,7 @@ where
     prometheus::set_score(paged_raw_solution.score);
 
     // 7. Submit the score and solution to the chain.
-    match dynamic::submit(&client, &signer, paged_raw_solution, listen)
+    match dynamic::submit(&client, &signer, paged_raw_solution, listen, chunk_size)
         .timed()
         .await
     {

src/commands/types.rs

@@ -185,4 +185,11 @@ pub struct ExperimentalMultiBlockMonitorConfig {
     /// Reduce the solution to prevent further trimming.
     #[clap(long, default_value_t = false)]
     pub do_reduce: bool,
+
+    /// Chunk size for submitting solution pages. If not specified or equal to zero,
+    /// all pages will be submitted concurrently. Otherwise, pages will be submitted in chunks
+    /// of the specified size, waiting for each chunk to be included in a block before
+    /// submitting the next chunk.
+    #[clap(long, default_value_t = 0)]
+    pub chunk_size: usize,
 }

src/dynamic/multi_block.rs

@@ -335,7 +335,12 @@ pub(crate) async fn submit<T: MinerConfig + Send + Sync + 'static>(
     signer: &Signer,
     mut paged_raw_solution: PagedRawSolution<T>,
     listen: Listen,
-) -> Result<(), Error> {
+    chunk_size: usize,
+) -> Result<(), Error>
+where
+    T::Solution: Send + Sync + 'static,
+    T::Pages: Send + Sync + 'static,
+{
     let mut i = 0;
     let tx_status = loop {
         let nonce = client
@@ -384,8 +389,14 @@ pub(crate) async fn submit<T: MinerConfig + Send + Sync + 'static>(
         .map(|(page, solution)| (page as u32, solution.clone()))
         .collect::<Vec<_>>();
 
-    // 2. Submit all solution pages.
-    let failed_pages = inner_submit_pages::<T>(client, signer, solutions, listen).await?;
+    // 2. Submit all solution pages using the appropriate strategy based on chunk_size
+    let failed_pages = if chunk_size == 0 {
+        // Use fully concurrent submission
+        inner_submit_pages_concurrent::<T>(client, signer, solutions, listen).await?
+    } else {
+        // Use chunked concurrent submission
+        inner_submit_pages_chunked::<T>(client, signer, solutions, listen, chunk_size).await?
+    };
 
     // 3. All pages were submitted successfully, we are done.
     if failed_pages.is_empty() {
@@ -405,7 +416,12 @@ pub(crate) async fn submit<T: MinerConfig + Send + Sync + 'static>(
         solutions.push((page, solution));
     }
 
-    let failed_pages = inner_submit_pages::<T>(client, signer, solutions, listen).await?;
+    // Retry with the same strategy as the initial submission
+    let failed_pages = if chunk_size == 0 {
+        inner_submit_pages_concurrent::<T>(client, signer, solutions, listen).await?
+    } else {
+        inner_submit_pages_chunked::<T>(client, signer, solutions, listen, chunk_size).await?
+    };
 
     if failed_pages.is_empty() {
         Ok(())
@@ -414,12 +430,17 @@ pub(crate) async fn submit<T: MinerConfig + Send + Sync + 'static>(
     }
 }
 
-pub(crate) async fn inner_submit_pages<T: MinerConfig + Send + Sync + 'static>(
+/// Submit all solution pages concurrently.
+pub(crate) async fn inner_submit_pages_concurrent<T: MinerConfig + Send + Sync + 'static>(
     client: &Client,
     signer: &Signer,
     paged_raw_solution: Vec<(u32, T::Solution)>,
     listen: Listen,
-) -> Result<Vec<u32>, Error> {
+) -> Result<Vec<u32>, Error>
+where
+    T::Solution: Send + Sync + 'static,
+    T::Pages: Send + Sync + 'static,
+{
     let mut txs = FuturesUnordered::new();
 
     let mut nonce = client
@@ -491,3 +512,99 @@ pub(crate) async fn inner_submit_pages<T: MinerConfig + Send + Sync + 'static>(
 
     Ok(failed_pages)
 }
+
+/// Submit solution pages in chunks, waiting for each chunk to be included in a block
+/// before submitting the next chunk.
+pub(crate) async fn inner_submit_pages_chunked<T: MinerConfig + Send + Sync + 'static>(
+    client: &Client,
+    signer: &Signer,
+    paged_raw_solution: Vec<(u32, T::Solution)>,
+    listen: Listen,
+    chunk_size: usize,
+) -> Result<Vec<u32>, Error>
+where
+    T::Solution: Send + Sync + 'static,
+    T::Pages: Send + Sync + 'static,
+{
+    let mut failed_pages = Vec::new();
+    let mut submitted_pages = HashSet::new();
+    let total_pages = paged_raw_solution.len();
+
+    // Process pages in chunks
+    for chunk_start in (0..total_pages).step_by(chunk_size) {
+        let chunk_end = std::cmp::min(chunk_start + chunk_size, total_pages);
+        let chunk = &paged_raw_solution[chunk_start..chunk_end];
+
+        log::info!(
+            target: LOG_TARGET,
+            "Submitting chunk of pages {}-{} of {}",
+            chunk_start,
+            chunk_end - 1,
+            total_pages
+        );
+
+        let mut txs = FuturesUnordered::new();
+        let mut nonce = client
+            .rpc()
+            .system_account_next_index(signer.account_id())
+            .await?;
+
+        // Submit all pages in the current chunk
+        for (page, solution) in chunk.iter() {
+            let tx_status = submit_inner(
+                client,
+                signer.clone(),
+                MultiBlockTransaction::submit_page::<T>(*page, Some(solution.clone()))?,
+                nonce,
+            )
+            .await?;
+
+            txs.push(async move {
+                match utils::wait_tx_in_block_for_strategy(tx_status, listen).await {
+                    Ok(tx) => Ok(tx),
+                    Err(_) => Err(*page),
+                }
+            });
+
+            nonce += 1;
+        }
+
+        // Wait for all pages in the current chunk to be included in a block
+        while let Some(page) = txs.next().await {
+            match page {
+                Ok(tx) => {
+                    let hash = tx.block_hash();
+
+                    let events = tx.wait_for_success().await?;
+                    for event in events.iter() {
+                        let event = event?;
+
+                        if let Some(solution_stored) =
+                            event.as_event::<runtime::multi_block_signed::events::Stored>()?
+                        {
+                            let page = solution_stored.2;
+
+                            log::info!(
+                                target: LOG_TARGET,
+                                "Page {page} included in block {:?}",
+                                hash
+                            );
+
+                            submitted_pages.insert(solution_stored.2);
+                        }
+                    }
+                }
+                Err(p) => {
+                    failed_pages.push(p);
+                }
+            }
+        }
+
+        // If all pages have been submitted, we're done
+        if submitted_pages.len() == total_pages {
+            return Ok(vec![]);
+        }
+    }
+
+    Ok(failed_pages)
+}

src/main.rs

@@ -397,6 +397,7 @@ mod tests {
                         listen: Listen::Finalized, // Assuming default
                         submission_strategy: SubmissionStrategy::IfLeading, // Assuming default
                         do_reduce: true,           // Expect true because flag was present
+                        chunk_size: 0,             // Default value
                     }
                 ),
             }
@@ -424,6 +425,35 @@ mod tests {
                     listen: Listen::Finalized,
                     submission_strategy: SubmissionStrategy::IfLeading,
                     do_reduce: false, // Expect false (default)
+                    chunk_size: 0,    // Default value
+                }
+            )
+        );
+    }
+
+    #[test]
+    fn cli_experimental_monitor_multi_block_with_chunk_size_works() {
+        let opt = Opt::try_parse_from([
+            env!("CARGO_PKG_NAME"),
+            "--uri",
+            "hi",
+            "experimental-monitor-multi-block",
+            "--seed-or-path",
+            "//Alice",
+            "--chunk-size",
+            "4",
+        ])
+        .unwrap();
+
+        assert_eq!(
+            opt.command,
+            Command::ExperimentalMonitorMultiBlock(
+                commands::types::ExperimentalMultiBlockMonitorConfig {
+                    seed_or_path: "//Alice".to_string(),
+                    listen: Listen::Finalized,
+                    submission_strategy: SubmissionStrategy::IfLeading,
+                    do_reduce: false, // Default value
+                    chunk_size: 4,    // Explicitly set to 4
                 }
             )
         );