Add pieces verification to farmers.

crates/sc-consensus-subspace-rpc/src/lib.rs

@@ -54,6 +54,7 @@ use subspace_core_primitives::{
 };
 use subspace_rpc_primitives::{
     FarmerProtocolInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
+    MAX_SEGMENT_INDEXES_PER_REQUEST,
 };
 
 const SOLUTION_TIMEOUT: Duration = Duration::from_secs(2);
@@ -527,6 +528,18 @@ where
     }
 
     async fn records_roots(&self, segment_indexes: Vec<u64>) -> RpcResult<Vec<Option<Sha256Hash>>> {
+        if segment_indexes.len() > MAX_SEGMENT_INDEXES_PER_REQUEST {
+            error!(
+                "segment_indexes length exceed the limit: {} ",
+                segment_indexes.len()
+            );
+
+            return Err(JsonRpseeError::Custom(
+                "Internal error during `records_root` call: segment_indexes length exceed the limit"
+                    .to_string()
+            ));
+        };
+
         let runtime_api = self.client.runtime_api();
         let best_block_id = BlockId::Hash(self.client.info().best_hash);
 

crates/subspace-farmer/src/bench_rpc_client.rs

@@ -1,9 +1,9 @@
-use std::num::NonZeroU32;
 use crate::rpc_client::{Error, RpcClient};
 use crate::utils::AbortingJoinHandle;
 use async_trait::async_trait;
 use futures::channel::mpsc;
 use futures::{stream, SinkExt, Stream, StreamExt};
+use std::num::NonZeroU32;
 use std::pin::Pin;
 use std::sync::Arc;
 use subspace_archiving::archiver::ArchivedSegment;
@@ -33,7 +33,7 @@ pub const BENCH_FARMER_PROTOCOL_INFO: FarmerProtocolInfo = FarmerProtocolInfo {
     genesis_hash: [0; 32],
     // PIECE_SIZE - WITNESS_SIZE
     record_size: NonZeroU32::new(3840).expect("We must set non-zero integer here."),
-    recorded_history_segment_size: 491520,   // RECORD_SIZE * MERKLE_NUM_LEAVES / 2
+    recorded_history_segment_size: 491520, // RECORD_SIZE * MERKLE_NUM_LEAVES / 2
     max_plot_size: 100 * 1024 * 1024 * 1024, // 100G
     // Doesn't matter, as we don't start sync
     total_pieces: 0,

crates/subspace-farmer/src/lib.rs

@@ -5,7 +5,7 @@
     io_error_other,
     map_first_last,
     trait_alias,
-    try_blocks,
+    try_blocks
 )]
 
 //! # `subspace-farmer` library implementation overview

crates/subspace-farmer/src/single_plot_farm.rs

@@ -16,7 +16,7 @@ use crate::ws_rpc_server::PieceGetter;
 use anyhow::anyhow;
 use async_trait::async_trait;
 use derive_more::{Display, From};
-use futures::future::try_join;
+use futures::future::{try_join, try_join_all};
 use num_traits::Zero;
 use serde::{Deserialize, Serialize};
 use std::future::Future;
@@ -34,7 +34,7 @@ use subspace_networking::{
     Config, Node, NodeRunner, PiecesByRangeRequest, PiecesByRangeRequestHandler,
     PiecesByRangeResponse, PiecesToPlot,
 };
-use subspace_rpc_primitives::FarmerProtocolInfo;
+use subspace_rpc_primitives::{FarmerProtocolInfo, MAX_SEGMENT_INDEXES_PER_REQUEST};
 use subspace_solving::{BatchEncodeError, SubspaceCodec};
 use thiserror::Error;
 use tokio::runtime::Handle;
@@ -752,19 +752,29 @@ impl<RC: RpcClient> VerifyingPlotter<RC> {
             return Err(PiecesVerificationError::InvalidFarmerProtocolInfo);
         }
 
+        // Calculate segment indexes collection
         let segment_indexes = piece_indexes
             .iter()
             .map(|piece_index| piece_index / records_per_segment as u64)
             .collect::<Vec<_>>();
 
-        let roots = self
-            .verification_client
-            .records_roots(segment_indexes)
-            .await
-            .map_err(|err| PiecesVerificationError::RpcError(err))?;
+        // Split segment indexes collection into allowed max sized chunks
+        // and run a future records_roots RPC call for each chunk.
+        let roots_futures = segment_indexes
+            .chunks(MAX_SEGMENT_INDEXES_PER_REQUEST)
+            .map(|segment_indexes_chunk| {
+                self.verification_client
+                    .records_roots(segment_indexes_chunk.to_vec())
+            })
+            .collect::<Vec<_>>();
 
-        let verified_roots = roots
+        // Wait for all the RPC calls,  flatten the results collection
+        // and check for empty result for any of the records root call.
+        let roots = try_join_all(roots_futures)
+            .await
+            .map_err(|err| PiecesVerificationError::RpcError(err))?
             .into_iter()
+            .flatten()
             .zip(piece_indexes.iter())
             .map(|(root, piece_index)| {
                 if let Some(root) = root {
@@ -777,8 +787,9 @@ impl<RC: RpcClient> VerifyingPlotter<RC> {
             })
             .collect::<Result<Vec<Sha256Hash>, PiecesVerificationError>>()?;
 
+        // Perform an actual piece validity check
         for ((piece, piece_index), root) in
-            pieces.as_pieces().zip(piece_indexes).zip(verified_roots)
+            pieces.as_pieces().zip(piece_indexes).zip(roots)
         {
             let position: u64 = piece_index % records_per_segment as u64;
 

crates/subspace-rpc-primitives/src/lib.rs

@@ -15,12 +15,15 @@
 
 //! Primitives for Subspace RPC.
 
-use std::num::NonZeroU32;
 use serde::{Deserialize, Serialize};
+use std::num::NonZeroU32;
 use subspace_core_primitives::{
     PublicKey, RewardSignature, Salt, Sha256Hash, SlotNumber, Solution,
 };
 
+/// Defines a limit for segment indexes array. It affects storage access on the runtime side.
+pub const MAX_SEGMENT_INDEXES_PER_REQUEST: usize = 300;
+
 /// Information about the protocol necessary for farmer operation
 #[derive(Debug, Copy, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]