Add pieces verification to farmers.

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

@@ -48,7 +48,9 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use subspace_archiving::archiver::ArchivedSegment;
-use subspace_core_primitives::{Solution, PIECE_SIZE, RECORDED_HISTORY_SEGMENT_SIZE, RECORD_SIZE};
+use subspace_core_primitives::{
+    Sha256Hash, Solution, PIECE_SIZE, RECORDED_HISTORY_SEGMENT_SIZE, RECORD_SIZE,
+};
 use subspace_rpc_primitives::{
     FarmerProtocolInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
 };
@@ -95,6 +97,9 @@ pub trait SubspaceRpcApi {
 
     #[method(name = "subspace_acknowledgeArchivedSegment")]
     async fn acknowledge_archived_segment(&self, segment_index: u64) -> RpcResult<()>;
+
+    #[method(name = "subspace_recordsRoot")]
+    async fn records_root(&self, segment_index: u64) -> RpcResult<Option<Sha256Hash>>;
 }
 
 #[derive(Default)]
@@ -513,4 +518,19 @@ where
 
         Ok(())
     }
+
+    async fn records_root(&self, segment_index: u64) -> RpcResult<Option<Sha256Hash>> {
+        let runtime_api = self.client.runtime_api();
+        let best_block_id = BlockId::Hash(self.client.info().best_hash);
+
+        let records_root = runtime_api.records_root(&best_block_id, segment_index);
+
+        records_root.map_err(|error| {
+            error!(
+                "Failed to get data from runtime API (records_root): {}",
+                error
+            );
+            JsonRpseeError::Custom("Internal error during `records_root` call".to_string())
+        })
+    }
 }

crates/subspace-farmer/src/bench_rpc_client.rs

@@ -6,6 +6,7 @@ use futures::{stream, SinkExt, Stream, StreamExt};
 use std::pin::Pin;
 use std::sync::Arc;
 use subspace_archiving::archiver::ArchivedSegment;
+use subspace_core_primitives::Sha256Hash;
 use subspace_rpc_primitives::{
     FarmerProtocolInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
 };
@@ -137,4 +138,8 @@ impl RpcClient for BenchRpcClient {
             .await?;
         Ok(())
     }
+
+    async fn records_root(&self, _: u64) -> Result<Option<Sha256Hash>, Error> {
+        Ok(None)
+    }
 }

crates/subspace-farmer/src/dsn.rs

@@ -1,3 +1,4 @@
+use async_trait::async_trait;
 use futures::{SinkExt, Stream, StreamExt};
 use num_traits::{WrappingAdd, WrappingSub};
 use std::ops::Range;
@@ -162,16 +163,23 @@ impl DSNSync for subspace_networking::Node {
     }
 }
 
+/// Defines actions on receiving the pieces during the sync process.
+#[async_trait]
+pub trait OnSync {
+    /// Defines a callback on receiving pieces.
+    async fn on_pieces(
+        &self,
+        pieces: FlatPieces,
+        piece_indices: Vec<PieceIndex>,
+    ) -> anyhow::Result<()>;
+}
+
 /// Syncs the closest pieces to the public key from the provided DSN.
-pub async fn sync<DSN, OP>(
-    mut dsn: DSN,
-    options: SyncOptions,
-    mut on_pieces: OP,
-) -> anyhow::Result<()>
+pub async fn sync<DSN, OP>(mut dsn: DSN, options: SyncOptions, on_sync: OP) -> anyhow::Result<()>
 where
     DSN: DSNSync + Send + Sized,
     DSN::Stream: Unpin + Send,
-    OP: FnMut(FlatPieces, Vec<PieceIndex>) -> anyhow::Result<()> + Send + 'static,
+    OP: OnSync,
 {
     let SyncOptions {
         max_plot_size,
@@ -232,11 +240,12 @@ where
     for (start, end) in sync_ranges {
         let mut stream = dsn.get_pieces(start.into()..end.into()).await?;
 
-        while let Some(PiecesToPlot {
-            piece_indexes,
-            pieces,
-        }) = stream.next().await
-        {
+        while let Some(pieces_to_plot) = stream.next().await {
+            let PiecesToPlot {
+                piece_indexes,
+                pieces,
+            } = pieces_to_plot;
+
             // Filter out pieces which are not in our range
             let (piece_indexes, pieces) = piece_indexes
                 .into_iter()
@@ -256,12 +265,10 @@ where
                 })
                 .unzip();
 
-            // Writing pieces is usually synchronous, therefore might take some time
-            on_pieces = tokio::task::spawn_blocking(move || {
-                on_pieces(pieces, piece_indexes).map(|()| on_pieces)
-            })
-            .await
-            .expect("`on_pieces` must never panic")?;
+            on_sync
+                .on_pieces(pieces, piece_indexes)
+                .await
+                .expect("`on_pieces` must never panic");
         }
     }
 

crates/subspace-farmer/src/dsn/tests.rs

@@ -1,5 +1,6 @@
 use super::{sync, DSNSync, NoSync, PieceIndexHashNumber, PiecesToPlot, SyncOptions};
 use crate::bench_rpc_client::{BenchRpcClient, BENCH_FARMER_PROTOCOL_INFO};
+use crate::dsn::OnSync;
 use crate::legacy_multi_plots_farm::{LegacyMultiPlotsFarm, Options as MultiFarmingOptions};
 use crate::single_plot_farm::PlotFactoryOptions;
 use crate::{LegacyObjectMappings, Plot};
@@ -20,6 +21,29 @@ use subspace_core_primitives::{
 use subspace_networking::libp2p::multiaddr::Protocol;
 use tempfile::TempDir;
 
+struct TestPlotter {
+    pub(crate) result: Arc<Mutex<BTreeMap<PieceIndexHash, (Piece, PieceIndex)>>>,
+}
+
+#[async_trait::async_trait]
+impl OnSync for TestPlotter {
+    async fn on_pieces(
+        &self,
+        pieces: FlatPieces,
+        piece_indices: Vec<PieceIndex>,
+    ) -> anyhow::Result<()> {
+        let mut result = self.result.lock();
+        result.extend(pieces.as_pieces().zip(piece_indices).map(|(piece, index)| {
+            (
+                PieceIndexHash::from_index(index),
+                (piece.try_into().unwrap(), index),
+            )
+        }));
+
+        Ok(())
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
 struct TestDSN(BTreeMap<PieceIndexHash, (Piece, PieceIndex)>);
 
@@ -80,19 +104,8 @@ async fn simple_test() {
             max_plot_size: 100 * 1024 * 1024 * 1024,
             total_pieces: 256,
         },
-        {
-            let result = Arc::clone(&result);
-            move |pieces, piece_indexes| {
-                let mut result = result.lock();
-                result.extend(pieces.as_pieces().zip(piece_indexes).map(|(piece, index)| {
-                    (
-                        PieceIndexHash::from_index(index),
-                        (piece.try_into().unwrap(), index),
-                    )
-                }));
-
-                Ok(())
-            }
+        TestPlotter {
+            result: Arc::clone(&result),
         },
     )
     .await
@@ -117,18 +130,8 @@ async fn no_sync_test() {
             max_plot_size: 100 * 1024 * 1024 * 1024,
             total_pieces: 0,
         },
-        {
-            let result = Arc::clone(&result);
-            move |pieces, piece_indexes| {
-                let mut result = result.lock();
-                result.extend(pieces.as_pieces().zip(piece_indexes).map(|(piece, index)| {
-                    (
-                        PieceIndexHash::from_index(index),
-                        (piece.try_into().unwrap(), index),
-                    )
-                }));
-                Ok(())
-            }
+        TestPlotter {
+            result: Arc::clone(&result),
         },
     )
     .await
@@ -359,7 +362,8 @@ async fn test_dsn_sync() {
 
     let range_size = PieceIndexHashNumber::MAX / seeder_max_piece_count * pieces_per_request;
     let plot = syncer_multi_farming.single_plot_farms()[0].plot().clone();
-    let dsn_sync = syncer_multi_farming.single_plot_farms()[0].dsn_sync(
+    let dsn_sync = syncer_multi_farming.single_plot_farms()[0].dsn_sync::<BenchRpcClient>(
+        None,
         syncer_max_plot_size,
         seeder_max_piece_count,
         range_size,

crates/subspace-farmer/src/lib.rs

@@ -40,6 +40,7 @@ pub mod legacy_multi_plots_farm;
 mod mock_rpc_client;
 pub(crate) mod node_rpc_client;
 pub(crate) mod object_mappings;
+mod pieces_verification;
 pub(crate) mod plot;
 pub(crate) mod rpc_client;
 pub mod single_disk_farm;

crates/subspace-farmer/src/mock_rpc_client.rs

@@ -5,6 +5,7 @@ use futures::{SinkExt, Stream, StreamExt};
 use std::pin::Pin;
 use std::sync::Arc;
 use subspace_archiving::archiver::ArchivedSegment;
+use subspace_core_primitives::Sha256Hash;
 use subspace_rpc_primitives::{
     FarmerProtocolInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
 };
@@ -232,4 +233,8 @@ impl RpcClient for MockRpcClient {
             .unwrap();
         Ok(())
     }
+
+    async fn records_root(&self, _: u64) -> Result<Option<Sha256Hash>, MockError> {
+        Ok(None)
+    }
 }

crates/subspace-farmer/src/node_rpc_client.rs

@@ -8,6 +8,7 @@ use jsonrpsee::ws_client::{WsClient, WsClientBuilder};
 use std::pin::Pin;
 use std::sync::Arc;
 use subspace_archiving::archiver::ArchivedSegment;
+use subspace_core_primitives::Sha256Hash;
 use subspace_rpc_primitives::{
     FarmerProtocolInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
 };
@@ -122,4 +123,11 @@ impl RpcClient for NodeRpcClient {
             )
             .await?)
     }
+
+    async fn records_root(&self, segment_index: u64) -> Result<Option<Sha256Hash>, RpcError> {
+        Ok(self
+            .client
+            .request("subspace_recordsRoot", rpc_params![&segment_index])
+            .await?)
+    }
 }

crates/subspace-farmer/src/pieces_verification.rs

@@ -0,0 +1,53 @@
+use crate::RpcClient;
+use subspace_archiving::archiver::is_piece_valid;
+use subspace_core_primitives::{
+    FlatPieces, PieceIndex, MERKLE_NUM_LEAVES, PIECE_SIZE, RECORD_SIZE,
+};
+use thiserror::Error;
+
+/// Pieces verification errors.
+#[derive(Error, Debug)]
+pub enum PiecesVerificationError {
+    /// Invalid pieces data provided
+    #[error("Invalid pieces data provided")]
+    InvalidRawData,
+    /// Pieces verification failed.
+    #[error("Pieces verification failed.")]
+    InvalidPieces,
+    /// RPC client failed.
+    #[error("RPC client failed. jsonrpsee error: {0}")]
+    RpcError(Box<dyn std::error::Error + Send + Sync>),
+    /// RPC client returned empty records_root.
+    #[error("RPC client returned empty records_root.")]
+    NoRecordsRootFound,
+}
+
+//TODO: Optimize: parallel execution, batch execution.
+/// Verifies pieces against the blockchain.
+pub async fn verify_pieces_at_blockchain<RC: RpcClient>(
+    verification_client: &RC,
+    piece_indexes: &[PieceIndex],
+    pieces: &FlatPieces,
+) -> Result<(), PiecesVerificationError> {
+    if piece_indexes.len() != (pieces.len() / PIECE_SIZE) {
+        return Err(PiecesVerificationError::InvalidRawData);
+    }
+    for (index, piece) in pieces.as_pieces().enumerate() {
+        let piece_index = piece_indexes[index];
+
+        let segment_index: u64 = piece_index / MERKLE_NUM_LEAVES as u64;
+        let position: u64 = piece_index % MERKLE_NUM_LEAVES as u64;
+
+        let root = verification_client
+            .records_root(segment_index)
+            .await
+            .map_err(|err| PiecesVerificationError::RpcError(err))?
+            .ok_or(PiecesVerificationError::NoRecordsRootFound)?;
+
+        if !is_piece_valid(piece, root, position as usize, RECORD_SIZE as usize) {
+            return Err(PiecesVerificationError::InvalidPieces);
+        }
+    }
+
+    Ok(())
+}

crates/subspace-farmer/src/rpc_client.rs

@@ -2,6 +2,7 @@ use async_trait::async_trait;
 use futures::Stream;
 use std::pin::Pin;
 use subspace_archiving::archiver::ArchivedSegment;
+use subspace_core_primitives::Sha256Hash;
 use subspace_rpc_primitives::{
     FarmerProtocolInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
 };
@@ -44,4 +45,7 @@ pub trait RpcClient: Clone + Send + Sync + 'static {
 
     /// Acknowledge receiving of archived segments
     async fn acknowledge_archived_segment(&self, segment_index: u64) -> Result<(), Error>;
+
+    /// Get records root for the segment
+    async fn records_root(&self, segment_index: u64) -> Result<Option<Sha256Hash>, Error>;
 }