Limit reconstruction concurrency on farmer to avoid excessive memory …

…usage

crates/subspace-farmer-components/src/plotting.rs

@@ -26,6 +26,7 @@ use subspace_core_primitives::{
 use subspace_erasure_coding::ErasureCoding;
 use subspace_proof_of_space::{Quality, Table};
 use thiserror::Error;
+use tokio::sync::Semaphore;
 use tracing::{debug, warn};
 
 fn default_backoff() -> ExponentialBackoff {
@@ -408,6 +409,10 @@ async fn download_sector<PG: PieceGetter>(
     piece_indexes: &[PieceIndex],
     piece_memory_cache: PieceMemoryCache,
 ) -> Result<(), PlottingError> {
+    // TODO: Make configurable, likely allowing user to specify RAM usage expectations and inferring
+    //  concurrency from there
+    let recovery_semaphore = Semaphore::new(1);
+
     let mut pieces_receiving_futures = piece_indexes
         .iter()
         .map(|piece_index| async {
@@ -428,6 +433,15 @@ async fn download_sector<PG: PieceGetter>(
 
             // all retries failed
             if !succeeded {
+                let _permit = match recovery_semaphore.acquire().await {
+                    Ok(permit) => permit,
+                    Err(error) => {
+                        return (
+                            piece_index,
+                            Err(format!("Recovery semaphore was closed: {error}").into()),
+                        );
+                    }
+                };
                 let recovered_piece =
                     recover_missing_piece(piece_getter, kzg.clone(), piece_index).await;
 

crates/subspace-farmer-components/src/segment_reconstruction.rs

@@ -9,6 +9,7 @@ use thiserror::Error;
 use tokio::sync::Semaphore;
 use tracing::{debug, error, info, trace, warn};
 
+// TODO: Probably should be made configurable
 const PARALLELISM_LEVEL: usize = 20;
 
 #[derive(Debug, Error)]