Merge pull request paritytech#458 from subspace/remove-rpc-workarounds

Remove farmer workarounds for node's RPC

Author: nazar-pc

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

@@ -47,7 +47,7 @@ use std::marker::PhantomData;
 use std::sync::Arc;
 use std::time::Duration;
 use subspace_archiving::archiver::ArchivedSegment;
-use subspace_core_primitives::{BlockNumber, Solution};
+use subspace_core_primitives::Solution;
 use subspace_rpc_primitives::{
     BlockSignature, BlockSigningInfo, FarmerMetadata, SlotInfo, SolutionResponse,
 };
@@ -61,10 +61,6 @@ pub trait SubspaceRpcApi {
     #[method(name = "subspace_getFarmerMetadata")]
     fn get_farmer_metadata(&self) -> RpcResult<FarmerMetadata>;
 
-    /// Get best block number
-    #[method(name = "subspace_getBestBlockNumber")]
-    fn get_best_block_number(&self) -> RpcResult<BlockNumber>;
-
     #[method(name = "subspace_submitSolutionResponse")]
     fn submit_solution_response(&self, solution_response: SolutionResponse) -> RpcResult<()>;
 
@@ -209,15 +205,6 @@ where
         })
     }
 
-    fn get_best_block_number(&self) -> RpcResult<BlockNumber> {
-        let best_number = TryInto::<BlockNumber>::try_into(self.client.info().best_number)
-            .unwrap_or_else(|_| {
-                panic!("Block number can't be converted into BlockNumber");
-            });
-
-        Ok(best_number)
-    }
-
     fn submit_solution_response(&self, solution_response: SolutionResponse) -> RpcResult<()> {
         let solution_response_senders = self.solution_response_senders.clone();
 

crates/subspace-farmer/src/archiving.rs

@@ -1,17 +1,13 @@
 use crate::object_mappings::ObjectMappings;
 use crate::rpc_client::RpcClient;
-use futures::{SinkExt, StreamExt};
-use std::time::Duration;
 use subspace_archiving::archiver::ArchivedSegment;
 use subspace_core_primitives::objects::{GlobalObject, PieceObject, PieceObjectMapping};
 use subspace_core_primitives::{FlatPieces, Sha256Hash};
 use subspace_rpc_primitives::FarmerMetadata;
 use thiserror::Error;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
-use tracing::{debug, error, info, warn};
-
-const BEST_BLOCK_REQUEST_TIMEOUT: Duration = Duration::from_secs(60);
+use tracing::{debug, error, info};
 
 #[derive(Debug, Error)]
 pub enum ArchivingError {
@@ -52,7 +48,6 @@ impl Archiving {
         farmer_metadata: FarmerMetadata,
         object_mappings: ObjectMappings,
         client: Client,
-        best_block_number_check_interval: Duration,
         mut on_pieces_to_plot: OPTP,
     ) -> Result<Archiving, ArchivingError>
     where
@@ -126,38 +121,6 @@ impl Archiving {
             .await
             .map_err(ArchivingError::RpcError)?;
 
-        let (mut best_block_number_sender, mut best_block_number_receiver) =
-            futures::channel::mpsc::channel(1);
-
-        tokio::spawn({
-            let client = client.clone();
-
-            async move {
-                loop {
-                    tokio::time::sleep(best_block_number_check_interval).await;
-
-                    // In case connection dies, we need to disconnect from the node
-                    let best_block_number_result = tokio::time::timeout(
-                        BEST_BLOCK_REQUEST_TIMEOUT,
-                        client.best_block_number(),
-                    )
-                    .await;
-
-                    let is_error = !matches!(best_block_number_result, Ok(Ok(_)));
-                    // Result doesn't matter here
-                    let _ = best_block_number_sender
-                        .send(best_block_number_result)
-                        .await;
-
-                    if is_error {
-                        break;
-                    }
-                }
-            }
-        });
-
-        let mut last_best_block_number_error = false;
-
         let archiving_handle = tokio::spawn(async move {
             // Listen for new blocks produced on the network
             loop {
@@ -189,36 +152,6 @@ impl Archiving {
                             }
                         }
                     }
-                    maybe_result = best_block_number_receiver.next() => {
-                        match maybe_result {
-                            Some(Ok(Ok(best_block_number))) => {
-                                debug!(best_block_number);
-                                last_best_block_number_error = false;
-                            }
-                            Some(Ok(Err(error))) => {
-                                if last_best_block_number_error {
-                                    error!(%error, "Request to get new best block failed second time");
-                                    break;
-                                } else {
-                                    warn!(%error, "Request to get new best block failed");
-                                    last_best_block_number_error = true;
-                                }
-                            }
-                            Some(Err(_error)) => {
-                                if last_best_block_number_error {
-                                    error!("Request to get new best block timed out second time");
-                                    break;
-                                } else {
-                                    warn!("Request to get new best block timed out");
-                                    last_best_block_number_error = true;
-                                }
-                            }
-                            None => {
-                                debug!("Best block number channel closed!");
-                                break;
-                            }
-                        }
-                    }
                 }
             }
         });

crates/subspace-farmer/src/bin/subspace-farmer/bench_rpc_client.rs

@@ -1,17 +1,13 @@
-use std::sync::Arc;
-
 use async_trait::async_trait;
+use std::sync::Arc;
 use subspace_archiving::archiver::ArchivedSegment;
-use subspace_core_primitives::BlockNumber;
+use subspace_farmer::{RpcClient, RpcClientError as MockError};
 use subspace_rpc_primitives::{
     BlockSignature, BlockSigningInfo, FarmerMetadata, SlotInfo, SolutionResponse,
 };
-use tokio::sync::mpsc::Receiver;
 use tokio::sync::{mpsc, Mutex};
 use tokio::task::JoinHandle;
 
-use subspace_farmer::{RpcClient, RpcClientError as MockError};
-
 /// Client mock for benching purpose
 #[derive(Clone, Debug)]
 pub struct BenchRpcClient {
@@ -70,11 +66,6 @@ impl RpcClient for BenchRpcClient {
         Ok(self.inner.metadata.clone())
     }
 
-    async fn best_block_number(&self) -> Result<BlockNumber, MockError> {
-        // Doesn't matter for tests (at least yet)
-        Ok(BlockNumber::MAX)
-    }
-
     async fn subscribe_slot_info(&self) -> Result<mpsc::Receiver<SlotInfo>, MockError> {
         unreachable!("Unreachable, as we don't start farming for benchmarking")
     }
@@ -86,7 +77,7 @@ impl RpcClient for BenchRpcClient {
         unreachable!("Unreachable, as we don't start farming for benchmarking")
     }
 
-    async fn subscribe_block_signing(&self) -> Result<Receiver<BlockSigningInfo>, MockError> {
+    async fn subscribe_block_signing(&self) -> Result<mpsc::Receiver<BlockSigningInfo>, MockError> {
         unreachable!("Unreachable, as we don't start farming for benchmarking")
     }
 
@@ -97,7 +88,9 @@ impl RpcClient for BenchRpcClient {
         unreachable!("Unreachable, as we don't start farming for benchmarking")
     }
 
-    async fn subscribe_archived_segments(&self) -> Result<Receiver<ArchivedSegment>, MockError> {
+    async fn subscribe_archived_segments(
+        &self,
+    ) -> Result<mpsc::Receiver<ArchivedSegment>, MockError> {
         let (sender, receiver) = mpsc::channel(10);
         let archived_segments_receiver = self.inner.archived_segments_receiver.clone();
         tokio::spawn(async move {

crates/subspace-farmer/src/bin/subspace-farmer/commands/farm.rs

@@ -3,7 +3,6 @@ use jsonrpsee::ws_server::WsServerBuilder;
 use rand::prelude::*;
 use std::path::PathBuf;
 use std::sync::Arc;
-use std::time::Duration;
 use std::{io, mem};
 use subspace_archiving::archiver::ArchivedSegment;
 use subspace_core_primitives::objects::{PieceObject, PieceObjectMapping};
@@ -93,7 +92,6 @@ pub(crate) async fn farm(
         plot_size,
         max_plot_size,
     }: FarmingArgs,
-    best_block_number_check_interval: Duration,
 ) -> Result<(), anyhow::Error> {
     raise_fd_limit();
 
@@ -136,7 +134,6 @@ pub(crate) async fn farm(
             client,
             object_mappings: object_mappings.clone(),
             reward_address,
-            best_block_number_check_interval,
         },
         plot_size,
         max_plot_size,
@@ -228,7 +225,6 @@ pub(crate) async fn bench(
     custom_path: Option<PathBuf>,
     plot_size: u64,
     max_plot_size: Option<u64>,
-    best_block_number_check_interval: Duration,
     write_to_disk: WriteToDisk,
     write_pieces_size: u64,
 ) -> anyhow::Result<()> {
@@ -281,7 +277,6 @@ pub(crate) async fn bench(
             client: client.clone(),
             object_mappings: object_mappings.clone(),
             reward_address: PublicKey::default(),
-            best_block_number_check_interval,
         },
         plot_size,
         max_plot_size,

crates/subspace-farmer/src/bin/subspace-farmer/main.rs

@@ -7,7 +7,6 @@ use clap::{ArgEnum, Parser, ValueHint};
 use sp_core::crypto::PublicError;
 use std::net::SocketAddr;
 use std::path::PathBuf;
-use std::time::Duration;
 use subspace_core_primitives::PublicKey;
 use subspace_networking::libp2p::Multiaddr;
 use tracing::info;
@@ -18,8 +17,6 @@ use tracing_subscriber::{
     EnvFilter,
 };
 
-const BEST_BLOCK_NUMBER_CHECK_INTERVAL: Duration = Duration::from_secs(5);
-
 /// Arguments for farmer
 #[derive(Debug, Parser)]
 struct FarmingArgs {
@@ -150,7 +147,7 @@ async fn main() -> Result<()> {
             info!("Done");
         }
         Command::Farm(args) => {
-            commands::farm(args, BEST_BLOCK_NUMBER_CHECK_INTERVAL).await?;
+            commands::farm(args).await?;
         }
         Command::Bench {
             custom_path,
@@ -163,7 +160,6 @@ async fn main() -> Result<()> {
                 custom_path,
                 plot_size,
                 max_plot_size,
-                BEST_BLOCK_NUMBER_CHECK_INTERVAL,
                 write_to_disk,
                 write_pieces_size,
             )

crates/subspace-farmer/src/mock_rpc_client.rs

@@ -2,11 +2,9 @@ use crate::rpc_client::{Error as MockError, RpcClient};
 use async_trait::async_trait;
 use std::sync::Arc;
 use subspace_archiving::archiver::ArchivedSegment;
-use subspace_core_primitives::BlockNumber;
 use subspace_rpc_primitives::{
     BlockSignature, BlockSigningInfo, FarmerMetadata, SlotInfo, SolutionResponse,
 };
-use tokio::sync::mpsc::Receiver;
 use tokio::sync::{mpsc, Mutex};
 
 /// `MockRpc` wrapper.
@@ -135,11 +133,6 @@ impl RpcClient for MockRpcClient {
         Ok(self.inner.metadata_receiver.lock().await.try_recv()?)
     }
 
-    async fn best_block_number(&self) -> Result<BlockNumber, MockError> {
-        // Doesn't matter for tests (at least yet)
-        Ok(0)
-    }
-
     async fn subscribe_slot_info(&self) -> Result<mpsc::Receiver<SlotInfo>, MockError> {
         let (sender, receiver) = mpsc::channel(10);
         let slot_receiver = self.inner.slot_info_receiver.clone();
@@ -164,7 +157,7 @@ impl RpcClient for MockRpcClient {
         Ok(())
     }
 
-    async fn subscribe_block_signing(&self) -> Result<Receiver<BlockSigningInfo>, MockError> {
+    async fn subscribe_block_signing(&self) -> Result<mpsc::Receiver<BlockSigningInfo>, MockError> {
         let (sender, receiver) = mpsc::channel(10);
         let block_signing_receiver = self.inner.block_signing_info_receiver.clone();
         tokio::spawn(async move {
@@ -188,7 +181,9 @@ impl RpcClient for MockRpcClient {
         Ok(())
     }
 
-    async fn subscribe_archived_segments(&self) -> Result<Receiver<ArchivedSegment>, MockError> {
+    async fn subscribe_archived_segments(
+        &self,
+    ) -> Result<mpsc::Receiver<ArchivedSegment>, MockError> {
         let (sender, receiver) = mpsc::channel(10);
         let archived_segments_receiver = self.inner.archived_segments_receiver.clone();
         tokio::spawn(async move {

crates/subspace-farmer/src/multi_farming.rs

@@ -4,7 +4,8 @@ use crate::{
 use anyhow::anyhow;
 use futures::stream::{FuturesUnordered, StreamExt};
 use rayon::prelude::*;
-use std::{path::PathBuf, sync::Arc, time::Duration};
+use std::path::PathBuf;
+use std::sync::Arc;
 use subspace_core_primitives::{PublicKey, PIECE_SIZE};
 use subspace_solving::SubspaceCodec;
 use tracing::info;
@@ -43,7 +44,6 @@ pub struct Options<C: RpcClient> {
     pub client: C,
     pub object_mappings: ObjectMappings,
     pub reward_address: PublicKey,
-    pub best_block_number_check_interval: Duration,
 }
 
 impl MultiFarming {
@@ -54,7 +54,6 @@ impl MultiFarming {
             client,
             object_mappings,
             reward_address,
-            best_block_number_check_interval,
         }: Options<C>,
         total_plot_size: u64,
         max_plot_size: u64,
@@ -125,34 +124,28 @@ impl MultiFarming {
             .map_err(|error| anyhow!(error))?;
 
         // Start archiving task
-        let archiving = Archiving::start(
-            farmer_metadata,
-            object_mappings,
-            client.clone(),
-            best_block_number_check_interval,
-            {
-                let mut on_pieces_to_plots = plots
-                    .iter()
-                    .zip(subspace_codecs)
-                    .zip(&commitments)
-                    .map(|((plot, subspace_codec), commitments)| {
-                        plotting::plot_pieces(subspace_codec, plot, commitments.clone())
+        let archiving = Archiving::start(farmer_metadata, object_mappings, client.clone(), {
+            let mut on_pieces_to_plots = plots
+                .iter()
+                .zip(subspace_codecs)
+                .zip(&commitments)
+                .map(|((plot, subspace_codec), commitments)| {
+                    plotting::plot_pieces(subspace_codec, plot, commitments.clone())
+                })
+                .collect::<Vec<_>>();
+
+            move |pieces_to_plot| {
+                on_pieces_to_plots
+                    .par_iter_mut()
+                    .map(|on_pieces_to_plot| {
+                        // TODO: It might be desirable to not clone it and instead pick just
+                        //  unnecessary pieces and copy pieces once since different plots will
+                        //  care about different pieces
+                        on_pieces_to_plot(pieces_to_plot.clone())
                     })
-                    .collect::<Vec<_>>();
-
-                move |pieces_to_plot| {
-                    on_pieces_to_plots
-                        .par_iter_mut()
-                        .map(|on_pieces_to_plot| {
-                            // TODO: It might be desirable to not clone it and instead pick just
-                            //  unnecessary pieces and copy pieces once since different plots will
-                            //  care about different pieces
-                            on_pieces_to_plot(pieces_to_plot.clone())
-                        })
-                        .reduce(|| true, |result, should_continue| result && should_continue)
-                }
-            },
-        )
+                    .reduce(|| true, |result, should_continue| result && should_continue)
+            }
+        })
         .await?;
 
         Ok(Self {