Merge pull request #1194 from subspace/improve-farmer-ux-cleanup

Improve farmer UX by printing useful details, improve logging format and other minor details

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

@@ -1,5 +1,6 @@
 mod farm;
 mod info;
+mod shared;
 
 pub(crate) use farm::farm_multi_disk;
 pub(crate) use info::info;

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

@@ -1,6 +1,7 @@
 mod dsn;
 
 use crate::commands::farm::dsn::{configure_dsn, start_announcements_processor};
+use crate::commands::shared::print_disk_farm_info;
 use crate::utils::{get_required_plot_space_with_overhead, shutdown_signal};
 use crate::{DiskFarm, FarmingArgs};
 use anyhow::{anyhow, Result};
@@ -55,11 +56,12 @@ pub(crate) async fn farm_multi_disk(
         disable_farming,
         mut dsn,
         max_concurrent_plots,
+        no_info: _,
     } = farming_args;
 
     let readers_and_pieces = Arc::new(Mutex::new(None));
 
-    info!("Connecting to node RPC at {}", node_rpc_url);
+    info!(url = %node_rpc_url, "Connecting to node RPC");
     let node_client = NodeRpcClient::new(&node_rpc_url).await?;
     let concurrent_plotting_semaphore = Arc::new(tokio::sync::Semaphore::new(
         farming_args.max_concurrent_plots.get(),
@@ -127,7 +129,7 @@ pub(crate) async fn farm_multi_disk(
 
     // TODO: Check plot and metadata sizes to ensure there is enough space for farmer to not
     //  fail later
-    for disk_farm in disk_farms {
+    for (disk_farm_index, disk_farm) in disk_farms.into_iter().enumerate() {
         let minimum_plot_size = get_required_plot_space_with_overhead(PLOT_SECTOR_SIZE);
 
         if disk_farm.allocated_plotting_space < minimum_plot_size {
@@ -138,22 +140,29 @@ pub(crate) async fn farm_multi_disk(
             ));
         }
 
-        info!("Connecting to node RPC at {}", node_rpc_url);
+        debug!(url = %node_rpc_url, %disk_farm_index, "Connecting to node RPC");
         let node_client = NodeRpcClient::new(&node_rpc_url).await?;
 
-        let single_disk_plot_fut = SingleDiskPlot::new(SingleDiskPlotOptions {
-            directory: disk_farm.directory,
-            allocated_space: disk_farm.allocated_plotting_space,
-            node_client,
-            reward_address,
-            kzg: kzg.clone(),
-            piece_getter: piece_getter.clone(),
-            concurrent_plotting_semaphore: Arc::clone(&concurrent_plotting_semaphore),
-            piece_memory_cache: piece_memory_cache.clone(),
-        });
+        let single_disk_plot_fut = SingleDiskPlot::new(
+            SingleDiskPlotOptions {
+                directory: disk_farm.directory.clone(),
+                allocated_space: disk_farm.allocated_plotting_space,
+                node_client,
+                reward_address,
+                kzg: kzg.clone(),
+                piece_getter: piece_getter.clone(),
+                concurrent_plotting_semaphore: Arc::clone(&concurrent_plotting_semaphore),
+                piece_memory_cache: piece_memory_cache.clone(),
+            },
+            disk_farm_index,
+        );
 
         let single_disk_plot = single_disk_plot_fut.await?;
 
+        if !farming_args.no_info {
+            print_disk_farm_info(disk_farm.directory, disk_farm_index);
+        }
+
         single_disk_plots.push(single_disk_plot);
     }
 
@@ -163,7 +172,7 @@ pub(crate) async fn farm_multi_disk(
         .map(|single_disk_plot| single_disk_plot.piece_reader())
         .collect::<Vec<_>>();
 
-    debug!("Collecting already plotted pieces");
+    info!("Collecting already plotted pieces (this will take some time)...");
 
     // Collect already plotted pieces
     let plotted_pieces: HashMap<PieceIndexHash, PieceDetails> = single_disk_plots
@@ -205,7 +214,7 @@ pub(crate) async fn farm_multi_disk(
         // We implicitly ignore duplicates here, reading just from one of the plots
         .collect();
 
-    debug!("Finished collecting already plotted pieces");
+    info!("Finished collecting already plotted pieces successfully");
 
     readers_and_pieces
         .lock()
@@ -225,9 +234,10 @@ pub(crate) async fn farm_multi_disk(
             // Collect newly plotted pieces
             // TODO: Once we have replotting, this will have to be updated
             single_disk_plot
-                .on_sector_plotted(Arc::new(move |(plotted_sector, plotting_permit)| {
+                .on_sector_plotted(Arc::new(move |(sector_offset, plotted_sector, plotting_permit)| {
                     let plotting_permit = Arc::clone(plotting_permit);
                     let node = node.clone();
+                    let sector_offset = *sector_offset;
                     let sector_index = plotted_sector.sector_index;
 
                     let mut dropped_receiver = dropped_sender.subscribe();
@@ -289,7 +299,7 @@ pub(crate) async fn farm_multi_disk(
                             // Nothing is needed here, just driving all futures to completion
                         }
 
-                        info!(?sector_index, "Sector publishing was successful.");
+                        info!(%sector_offset, ?sector_index, "Sector publishing was successful.");
 
                         // Release only after publishing is finished
                         drop(plotting_permit);

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

@@ -39,6 +39,7 @@ pub(super) fn configure_dsn(
         listen_on,
         bootstrap_nodes,
         piece_cache_size,
+        provided_keys_limit,
         disable_private_ips,
         reserved_peers,
     }: DsnArgs,
@@ -63,33 +64,48 @@ pub(super) fn configure_dsn(
             fs::rename(&records_cache_db_path, &piece_cache_db_path)?;
         }
     }
-    let provider_cache_db_path = base_path.join("provider_cache_db");
-    let provider_cache_size =
-        piece_cache_size.saturating_mul(NonZeroUsize::new(10).expect("10 > 0")); // TODO: add proper value
-
-    info!(
-        ?piece_cache_db_path,
-        ?piece_cache_size,
-        ?provider_cache_db_path,
-        ?provider_cache_size,
-        "Record cache DB configured."
-    );
+    let provider_db_path = base_path.join("providers_db");
+    // TODO: Remove this migration code in the future
+    {
+        let provider_cache_db_path = base_path.join("provider_cache_db");
+        if provider_cache_db_path.exists() {
+            fs::rename(&provider_cache_db_path, &provider_db_path)?;
+        }
+    }
 
     let default_config = Config::default();
     let peer_id = peer_id(&keypair);
 
-    let db_provider_storage =
-        ParityDbProviderStorage::new(&provider_cache_db_path, provider_cache_size, peer_id)
+    info!(
+        db_path = ?provider_db_path,
+        keys_limit = ?provided_keys_limit,
+        "Initializing provider storage..."
+    );
+    let persistent_provider_storage =
+        ParityDbProviderStorage::new(&provider_db_path, provided_keys_limit, peer_id)
             .map_err(|err| anyhow::anyhow!(err.to_string()))?;
+    info!(
+        current_size = ?persistent_provider_storage.size(),
+        "Provider storage initialized successfully"
+    );
 
+    info!(
+        db_path = ?piece_cache_db_path,
+        size = ?piece_cache_size,
+        "Initializing piece cache..."
+    );
     let piece_store =
         ParityDbStore::new(&piece_cache_db_path).map_err(|err| anyhow::anyhow!(err.to_string()))?;
     let piece_cache = FarmerPieceCache::new(piece_store.clone(), piece_cache_size, peer_id);
+    info!(
+        current_size = ?piece_cache.size(),
+        "Piece cache initialized successfully"
+    );
 
     let farmer_provider_storage = FarmerProviderStorage::new(
         peer_id,
         readers_and_pieces.clone(),
-        db_provider_storage,
+        persistent_provider_storage,
         piece_cache.clone(),
     );
 

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

@@ -1,5 +1,5 @@
+use crate::commands::shared::print_disk_farm_info;
 use crate::DiskFarm;
-use subspace_farmer::single_disk_plot::{SingleDiskPlot, SingleDiskPlotSummary};
 
 pub(crate) fn info(disk_farms: Vec<DiskFarm>) {
     for (disk_farm_index, disk_farm) in disk_farms.into_iter().enumerate() {
@@ -9,28 +9,6 @@ pub(crate) fn info(disk_farms: Vec<DiskFarm>) {
 
         let DiskFarm { directory, .. } = disk_farm;
 
-        println!("Single disk farm {disk_farm_index}:");
-        match SingleDiskPlot::collect_summary(directory) {
-            SingleDiskPlotSummary::Found { info, directory } => {
-                println!("  ID: {}", info.id());
-                println!("  Genesis hash: 0x{}", hex::encode(info.genesis_hash()));
-                println!("  Public key: 0x{}", hex::encode(info.public_key()));
-                println!("  First sector index: {}", info.first_sector_index());
-                println!(
-                    "  Allocated space: {} ({})",
-                    bytesize::to_string(info.allocated_space(), true),
-                    bytesize::to_string(info.allocated_space(), false)
-                );
-                println!("  Directory: {}", directory.display());
-            }
-            SingleDiskPlotSummary::NotFound { directory } => {
-                println!("  Plot directory: {}", directory.display());
-                println!("  No farm found here yet");
-            }
-            SingleDiskPlotSummary::Error { directory, error } => {
-                println!("  Ddirectory: {}", directory.display());
-                println!("  Failed to open farm info: {error}");
-            }
-        }
+        print_disk_farm_info(directory, disk_farm_index);
     }
 }

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

@@ -0,0 +1,28 @@
+use std::path::PathBuf;
+use subspace_farmer::single_disk_plot::{SingleDiskPlot, SingleDiskPlotSummary};
+
+pub(crate) fn print_disk_farm_info(directory: PathBuf, disk_farm_index: usize) {
+    println!("Single disk farm {disk_farm_index}:");
+    match SingleDiskPlot::collect_summary(directory) {
+        SingleDiskPlotSummary::Found { info, directory } => {
+            println!("  ID: {}", info.id());
+            println!("  Genesis hash: 0x{}", hex::encode(info.genesis_hash()));
+            println!("  Public key: 0x{}", hex::encode(info.public_key()));
+            println!("  First sector index: {}", info.first_sector_index());
+            println!(
+                "  Allocated space: {} ({})",
+                bytesize::to_string(info.allocated_space(), true),
+                bytesize::to_string(info.allocated_space(), false)
+            );
+            println!("  Directory: {}", directory.display());
+        }
+        SingleDiskPlotSummary::NotFound { directory } => {
+            println!("  Plot directory: {}", directory.display());
+            println!("  No farm found here yet");
+        }
+        SingleDiskPlotSummary::Error { directory, error } => {
+            println!("  Directory: {}", directory.display());
+            println!("  Failed to open farm info: {error}");
+        }
+    }
+}

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

@@ -19,7 +19,6 @@ use subspace_networking::libp2p::Multiaddr;
 use tempfile::TempDir;
 use tracing::info;
 use tracing_subscriber::filter::LevelFilter;
-use tracing_subscriber::fmt::format::FmtSpan;
 use tracing_subscriber::prelude::*;
 use tracing_subscriber::{fmt, EnvFilter};
 
@@ -56,6 +55,9 @@ struct FarmingArgs {
     /// Number of plots that can be plotted concurrently, impacts RAM usage.
     #[arg(long, default_value = "10")]
     max_concurrent_plots: NonZeroUsize,
+    /// Do not print info about configured farms on startup.
+    #[arg(long)]
+    no_info: bool,
 }
 
 /// Arguments for DSN
@@ -71,6 +73,9 @@ struct DsnArgs {
     /// Piece cache size in pieces.
     #[arg(long, default_value = "65536")]
     piece_cache_size: NonZeroUsize,
+    /// Number of provided keys (by other peers) that will be stored.
+    #[arg(long, default_value = "655360")]
+    provided_keys_limit: NonZeroUsize,
     /// Determines whether we allow keeping non-global (private, shared, loopback..) addresses in Kademlia DHT.
     #[arg(long, default_value_t = false)]
     disable_private_ips: bool,
@@ -202,7 +207,7 @@ struct Command {
 async fn main() -> Result<()> {
     tracing_subscriber::registry()
         .with(
-            fmt::layer().with_span_events(FmtSpan::CLOSE).with_filter(
+            fmt::layer().with_filter(
                 EnvFilter::builder()
                     .with_default_directive(LevelFilter::INFO.into())
                     .from_env_lossy(),