Add pieces verification to farmers.

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

@@ -49,7 +49,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use subspace_archiving::archiver::ArchivedSegment;
 use subspace_core_primitives::{
-    Solution, MERKLE_NUM_LEAVES, PIECE_SIZE, RECORDED_HISTORY_SEGMENT_SIZE, RECORD_SIZE,
+    Sha256Hash, Solution, MERKLE_NUM_LEAVES, PIECE_SIZE, RECORDED_HISTORY_SEGMENT_SIZE, RECORD_SIZE,
 };
 use subspace_rpc_primitives::{
     FarmerProtocolInfo, RewardSignatureResponse, RewardSigningInfo, SlotInfo, SolutionResponse,
@@ -97,6 +97,9 @@ pub trait SubspaceRpcApi {
 
     #[method(name = "subspace_acknowledgeArchivedSegment")]
     async fn acknowledge_archived_segment(&self, segment_index: u64) -> RpcResult<()>;
+
+    #[method(name = "subspace_recordsRoots")]
+    async fn records_roots(&self, segment_indexes: Vec<u64>) -> RpcResult<Vec<Option<Sha256Hash>>>;
 }
 
 #[derive(Default)]
@@ -518,4 +521,27 @@ where
 
         Ok(())
     }
+
+    async fn records_roots(&self, segment_indexes: Vec<u64>) -> RpcResult<Vec<Option<Sha256Hash>>> {
+        let runtime_api = self.client.runtime_api();
+        let best_block_id = BlockId::Hash(self.client.info().best_hash);
+
+        let records_root_result: Result<Vec<_>, JsonRpseeError> = segment_indexes
+            .into_iter()
+            .map(|idx| {
+                runtime_api.records_root(&best_block_id, idx).map_err(|_| {
+                    JsonRpseeError::Custom("Internal error during `records_root` call".to_string())
+                })
+            })
+            .collect();
+
+        if let Err(ref err) = records_root_result {
+            error!(
+                "Failed to get data from runtime API (records_root): {}",
+                err
+            );
+        }
+
+        records_root_result
+    }
 }

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_roots(&self, _: Vec<u64>) -> Result<Vec<Option<Sha256Hash>>, Error> {
+        Ok(Default::default())
+    }
 }

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;
@@ -6,7 +7,7 @@ use subspace_core_primitives::{
 };
 use subspace_networking::libp2p::core::multihash::{Code, MultihashDigest};
 use subspace_networking::{PiecesByRangeRequest, PiecesByRangeResponse, PiecesToPlot};
-use tracing::{debug, trace, warn};
+use tracing::{debug, error, trace, warn};
 
 #[cfg(test)]
 mod pieces_by_range_tests;
@@ -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,9 @@ 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")?;
+            if let Err(err) = on_sync.on_pieces(pieces, piece_indexes).await {
+                error!("DSN sync process returned an error: {}", err);
+            }
         }
     }
 

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};
@@ -15,11 +16,36 @@ use std::time::Duration;
 use subspace_archiving::archiver::ArchivedSegment;
 use subspace_core_primitives::{
     bidirectional_distance, ArchivedBlockProgress, FlatPieces, LastArchivedBlock, Piece,
-    PieceIndex, PieceIndexHash, RootBlock, Sha256Hash, PIECE_SIZE, U256,
+    PieceIndex, PieceIndexHash, RootBlock, Sha256Hash, PIECE_SIZE, RECORDED_HISTORY_SEGMENT_SIZE,
+    RECORD_SIZE, U256,
 };
 use subspace_networking::libp2p::multiaddr::Protocol;
+use subspace_rpc_primitives::FarmerProtocolInfo;
 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 +106,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 +132,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,10 +364,17 @@ 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(
-        syncer_max_plot_size,
-        seeder_max_piece_count,
+    let dsn_sync = syncer_multi_farming.single_plot_farms()[0].dsn_sync::<BenchRpcClient>(
+        rpc_client.clone(),
+        FarmerProtocolInfo {
+            max_plot_size: syncer_max_plot_size,
+            total_pieces: seeder_max_piece_count,
+            genesis_hash: Default::default(), // We don't use this field.
+            record_size: RECORD_SIZE,
+            recorded_history_segment_size: RECORDED_HISTORY_SEGMENT_SIZE,
+        },
         range_size,
+        false, // don't verify pieces
     );
     let public_key =
         U256::from_be_bytes((*syncer_multi_farming.single_plot_farms()[0].public_key()).into());

crates/subspace-farmer/src/legacy_multi_plots_farm.rs

@@ -78,6 +78,7 @@ impl LegacyMultiPlotsFarm {
         let single_disk_semaphore =
             SingleDiskSemaphore::new(NonZeroU16::try_from(16).expect("Non zero; qed"));
 
+        let verification_client = archiving_client.clone();
         let single_plot_farms = tokio::task::spawn_blocking(move || {
             let handle = Handle::current();
             plot_sizes
@@ -112,6 +113,7 @@ impl LegacyMultiPlotsFarm {
                         enable_dsn_archiving,
                         enable_dsn_sync,
                         relay_server_node: relay_server_node.clone(),
+                        verification_client: verification_client.clone(),
                     })
                 })
                 .collect::<anyhow::Result<Vec<_>>>()

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_roots(&self, _: Vec<u64>) -> Result<Vec<Option<Sha256Hash>>, MockError> {
+        Ok(Default::default())
+    }
 }

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,14 @@ impl RpcClient for NodeRpcClient {
             )
             .await?)
     }
+
+    async fn records_roots(
+        &self,
+        segment_indexes: Vec<u64>,
+    ) -> Result<Vec<Option<Sha256Hash>>, RpcError> {
+        Ok(self
+            .client
+            .request("subspace_recordsRoots", rpc_params![&segment_indexes])
+            .await?)
+    }
 }

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,10 @@ 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 roots for the segments
+    async fn records_roots(
+        &self,
+        segment_indexes: Vec<u64>,
+    ) -> Result<Vec<Option<Sha256Hash>>, Error>;
 }

crates/subspace-farmer/src/single_disk_farm.rs

@@ -313,7 +313,7 @@ impl SingleDiskFarm {
         };
 
         let single_disk_semaphore = SingleDiskSemaphore::new(disk_concurrency);
-
+        let verification_client = archiving_client.clone();
         let single_plot_farms = tokio::task::spawn_blocking(move || {
             let handle = Handle::current();
             single_disk_farm_info
@@ -350,6 +350,7 @@ impl SingleDiskFarm {
                         enable_dsn_archiving,
                         enable_dsn_sync,
                         relay_server_node: relay_server_node.clone(),
+                        verification_client: verification_client.clone(),
                     })
                 })
                 .collect::<anyhow::Result<Vec<_>>>()