Merge pull request #2865 from subspace/node-client-optimization

Node client optimization

.github/workflows/rustsec-audit.yml

@@ -20,5 +20,6 @@ jobs:
       - uses: rustsec/audit-check@dd51754d4e59da7395a4cd9b593f0ff2d61a9b95 #v1.4.1
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
-          # TODO: Remove once Substrate upgrades libp2p and we no longer have rustls 0.20.9 in our dependencies
-          ignore: RUSTSEC-2024-0336
+          # TODO: Remove once Substrate upgrades libp2p and we no longer have rustls 0.20.9 and curve25519-dalek 3.2.0
+          #  in our dependencies: https://github.com/paritytech/polkadot-sdk/pull/1631
+          ignore: RUSTSEC-2024-0336,RUSTSEC-2024-0344

crates/subspace-farmer/Cargo.toml

@@ -61,6 +61,7 @@ tempfile = "3.10.1"
 thiserror = "1.0.59"
 thread-priority = "1.1.0"
 tokio = { version = "1.38.0", features = ["macros", "parking_lot", "rt-multi-thread", "signal", "time"] }
+tokio-stream = { version = "0.1.15", features = ["sync"] }
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 ulid = { version = "1.1.2", features = ["serde"] }

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster/controller.rs

@@ -25,6 +25,7 @@ use subspace_farmer::farm::plotted_pieces::PlottedPieces;
 use subspace_farmer::farmer_cache::FarmerCache;
 use subspace_farmer::farmer_piece_getter::piece_validator::SegmentCommitmentPieceValidator;
 use subspace_farmer::farmer_piece_getter::{DsnCacheRetryPolicy, FarmerPieceGetter};
+use subspace_farmer::node_client::caching_proxy_node_client::CachingProxyNodeClient;
 use subspace_farmer::node_client::rpc_node_client::RpcNodeClient;
 use subspace_farmer::node_client::NodeClient;
 use subspace_farmer::single_disk_farm::identity::Identity;
@@ -159,6 +160,10 @@ pub(super) async fn controller(
         .map_err(|error| anyhow!("Failed to configure networking: {error}"))?
     };
 
+    let node_client = CachingProxyNodeClient::new(node_client)
+        .await
+        .map_err(|error| anyhow!("Failed to create caching proxy node client: {error}"))?;
+
     let kzg = Kzg::new(embedded_kzg_settings());
     let validator = Some(SegmentCommitmentPieceValidator::new(
         node.clone(),

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

@@ -24,6 +24,7 @@ use subspace_farmer::cluster::plotter::ClusterPlotter;
 use subspace_farmer::farm::{
     Farm, FarmingNotification, SectorExpirationDetails, SectorPlottingDetails, SectorUpdate,
 };
+use subspace_farmer::node_client::caching_proxy_node_client::CachingProxyNodeClient;
 use subspace_farmer::node_client::NodeClient;
 use subspace_farmer::single_disk_farm::{
     SingleDiskFarm, SingleDiskFarmError, SingleDiskFarmOptions,
@@ -171,9 +172,13 @@ where
         None
     };
 
-    let node_client = ClusterNodeClient::new(nats_client.clone())
-        .await
-        .map_err(|error| anyhow!("Failed to create cluster node client: {error}"))?;
+    let node_client = CachingProxyNodeClient::new(
+        ClusterNodeClient::new(nats_client.clone())
+            .await
+            .map_err(|error| anyhow!("Failed to create cluster node client: {error}"))?,
+    )
+    .await
+    .map_err(|error| anyhow!("Failed to create caching proxy node client: {error}"))?;
 
     let farmer_app_info = node_client
         .farmer_app_info()

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

@@ -30,6 +30,7 @@ use subspace_farmer::farm::{
 use subspace_farmer::farmer_cache::FarmerCache;
 use subspace_farmer::farmer_piece_getter::piece_validator::SegmentCommitmentPieceValidator;
 use subspace_farmer::farmer_piece_getter::{DsnCacheRetryPolicy, FarmerPieceGetter};
+use subspace_farmer::node_client::caching_proxy_node_client::CachingProxyNodeClient;
 use subspace_farmer::node_client::rpc_node_client::RpcNodeClient;
 use subspace_farmer::node_client::NodeClient;
 use subspace_farmer::plotter::cpu::CpuPlotter;
@@ -368,6 +369,10 @@ where
         .map_err(|error| anyhow!("Failed to configure networking: {error}"))?
     };
 
+    let node_client = CachingProxyNodeClient::new(node_client)
+        .await
+        .map_err(|error| anyhow!("Failed to create caching proxy node client: {error}"))?;
+
     let _prometheus_worker = if should_start_prometheus_server {
         let prometheus_task = start_prometheus_metrics_server(
             prometheus_listen_on,