removed the remainings of put batching

src/bin/avail-light.rs

@@ -138,7 +138,6 @@ async fn run(error_sender: Sender<anyhow::Error>) -> Result<()> {
 		(&cfg).into(),
 		cfg.dht_parallelization_limit,
 		cfg.kad_record_ttl,
-		cfg.put_batch_size,
 		cfg.is_fat_client(),
 		id_keys,
 	)

src/light_client.rs

@@ -36,7 +36,7 @@ use std::{
 	time::Instant,
 };
 use tokio::sync::{broadcast, mpsc::Sender};
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 use crate::{
 	data::{store_block_header_in_db, store_confidence_in_db},
@@ -53,8 +53,8 @@ use crate::{
 #[async_trait]
 #[automock]
 pub trait LightClient {
-	async fn insert_cells_into_dht(&self, block: u32, cells: Vec<Cell>) -> f32;
-	async fn insert_rows_into_dht(&self, block: u32, rows: Vec<(RowIndex, Vec<u8>)>) -> f32;
+	async fn insert_cells_into_dht(&self, block: u32, cells: Vec<Cell>) -> Result<()>;
+	async fn insert_rows_into_dht(&self, block: u32, rows: Vec<(RowIndex, Vec<u8>)>) -> Result<()>;
 	async fn get_kate_proof(&self, hash: H256, positions: &[Position]) -> Result<Vec<Cell>>;
 	async fn shrink_kademlia_map(&self) -> Result<()>;
 	async fn get_multiaddress_and_ip(&self) -> Result<(String, String)>;
@@ -80,13 +80,13 @@ pub fn new(db: Arc<DB>, p2p_client: P2pClient, rpc_client: RpcClient) -> impl Li
 
 #[async_trait]
 impl LightClient for LightClientImpl {
-	async fn insert_cells_into_dht(&self, block: u32, cells: Vec<Cell>) -> f32 {
+	async fn insert_cells_into_dht(&self, block: u32, cells: Vec<Cell>) -> Result<()> {
 		self.p2p_client.insert_cells_into_dht(block, cells).await
 	}
 	async fn shrink_kademlia_map(&self) -> Result<()> {
 		self.p2p_client.shrink_kademlia_map().await
 	}
-	async fn insert_rows_into_dht(&self, block: u32, rows: Vec<(RowIndex, Vec<u8>)>) -> f32 {
+	async fn insert_rows_into_dht(&self, block: u32, rows: Vec<(RowIndex, Vec<u8>)>) -> Result<()> {
 		self.p2p_client.insert_rows_into_dht(block, rows).await
 	}
 	async fn get_kate_proof(&self, hash: H256, positions: &[Position]) -> Result<Vec<Cell>> {
@@ -192,18 +192,6 @@ pub async fn process_block(
 			.await?;
 	}
 
-	if let Some(dht_put_success_rate) = fetch_stats.dht_put_success_rate {
-		metrics
-			.record(MetricValue::DHTPutSuccess(dht_put_success_rate))
-			.await?;
-	}
-
-	if let Some(dht_put_duration) = fetch_stats.dht_put_duration {
-		metrics
-			.record(MetricValue::DHTPutDuration(dht_put_duration))
-			.await?;
-	}
-
 	if positions.len() > fetched.len() {
 		error!(block_number, "Failed to fetch {} cells", unfetched.len());
 		return Ok(None);
@@ -305,22 +293,25 @@ pub async fn process_block(
 			.filter(|cell| !cell.position.is_extended())
 			.collect::<Vec<_>>();
 		let rpc_fetched_data_rows = data::rows(dimensions, &rpc_fetched_data_cells);
-		let rows_len = rpc_fetched_data_rows.len();
 
-		let dht_insert_rows_success_rate = light_client
+		light_client
 			.insert_rows_into_dht(block_number, rpc_fetched_data_rows)
-			.await;
-
-		info!(
-			block_number,
-			"DHT PUT rows operation success rate" = dht_insert_rows_success_rate,
-			"rows inserted into DHT" = rows_len
-		);
+			.await
+			.map_err(|e| {
+				warn!("Error inserting rows into DHT: {e}");
+				e
+			})
+			.ok();
 	}
 
-	_ = light_client
+	light_client
 		.insert_cells_into_dht(block_number, rpc_fetched)
-		.await;
+		.await
+		.map_err(|e| {
+			warn!("Error inserting cells into DHT: {e}");
+			e
+		})
+		.ok();
 
 	light_client
 		.shrink_kademlia_map()
@@ -567,7 +558,6 @@ mod tests {
 					fetched.len(),
 					Duration::from_secs(0),
 					None,
-					None,
 				);
 				Box::pin(async move { Ok((fetched, unfetched, stats)) })
 			});
@@ -583,10 +573,10 @@ mod tests {
 			.returning(|_, _| Ok(()));
 		mock_client
 			.expect_insert_rows_into_dht()
-			.returning(|_, _| Box::pin(async move { 1f32 }));
+			.returning(|_, _| Box::pin(async move { Ok(()) }));
 		mock_client
 			.expect_insert_cells_into_dht()
-			.returning(|_, _| Box::pin(async move { 1f32 }));
+			.returning(|_, _| Box::pin(async move { Ok(()) }));
 		mock_client
 			.expect_shrink_kademlia_map()
 			.returning(|| Box::pin(async move { Ok(()) }));
@@ -712,7 +702,6 @@ mod tests {
 					fetched.len(),
 					Duration::from_secs(0),
 					None,
-					None,
 				);
 				Box::pin(async move { Ok((fetched, unfetched, stats)) })
 			});
@@ -725,10 +714,10 @@ mod tests {
 			.returning(|_, _| Ok(()));
 		mock_client
 			.expect_insert_rows_into_dht()
-			.returning(|_, _| Box::pin(async move { 1f32 }));
+			.returning(|_, _| Box::pin(async move { Ok(()) }));
 		mock_client
 			.expect_insert_cells_into_dht()
-			.returning(|_, _| Box::pin(async move { 1f32 }));
+			.returning(|_, _| Box::pin(async move { Ok(()) }));
 		mock_client
 			.expect_shrink_kademlia_map()
 			.returning(|| Box::pin(async move { Ok(()) }));

src/network/mod.rs

@@ -10,7 +10,7 @@ use mockall::automock;
 use sp_core::H256;
 use std::{sync::Arc, time::Duration};
 use tokio::time::Instant;
-use tracing::info;
+use tracing::{info, warn};
 
 use crate::proof;
 
@@ -36,30 +36,23 @@ pub struct FetchStats {
 	pub dht_fetch_duration: f64,
 	pub rpc_fetched: Option<f64>,
 	pub rpc_fetch_duration: Option<f64>,
-	pub dht_put_success_rate: Option<f64>,
-	pub dht_put_duration: Option<f64>,
 }
 
 type RPCFetchStats = (usize, Duration);
 
-type DHTPutStats = (f64, Duration);
-
 impl FetchStats {
 	pub fn new(
 		total: usize,
 		dht_fetched: usize,
 		dht_fetch_duration: Duration,
 		rpc_fetch_stats: Option<RPCFetchStats>,
-		dht_put_stats: Option<DHTPutStats>,
 	) -> Self {
 		FetchStats {
 			dht_fetched: dht_fetched as f64,
 			dht_fetched_percentage: dht_fetched as f64 / total as f64,
 			dht_fetch_duration: dht_fetch_duration.as_secs_f64(),
 			rpc_fetched: rpc_fetch_stats.map(|(rpc_fetched, _)| rpc_fetched as f64),
 			rpc_fetch_duration: rpc_fetch_stats.map(|(_, duration)| duration.as_secs_f64()),
-			dht_put_success_rate: dht_put_stats.map(|(rate, _)| rate),
-			dht_put_duration: dht_put_stats.map(|(_, duration)| duration.as_secs_f64()),
 		}
 	}
 }
@@ -171,13 +164,8 @@ impl Client for DHTWithRPCFallbackClient {
 			.await?;
 
 		if self.disable_rpc {
-			let stats = FetchStats::new(
-				positions.len(),
-				dht_fetched.len(),
-				dht_fetch_duration,
-				None,
-				None,
-			);
+			let stats =
+				FetchStats::new(positions.len(), dht_fetched.len(), dht_fetch_duration, None);
 			return Ok((dht_fetched, unfetched, stats));
 		};
 
@@ -191,24 +179,20 @@ impl Client for DHTWithRPCFallbackClient {
 			)
 			.await?;
 
-		let begin = Instant::now();
-
-		let dht_put_success_rate = self
-			.p2p_client
+		self.p2p_client
 			.insert_cells_into_dht(block_number, rpc_fetched.clone())
-			.await;
-
-		info!(
-			block_number,
-			"DHT PUT operation success rate: {dht_put_success_rate}"
-		);
+			.await
+			.map_err(|e| {
+				warn!("Error inserting cells into DHT: {e}");
+				e
+			})
+			.ok();
 
 		let stats = FetchStats::new(
 			positions.len(),
 			dht_fetched.len(),
 			dht_fetch_duration,
 			Some((rpc_fetched.len(), rpc_fetch_duration)),
-			Some((dht_put_success_rate as f64, begin.elapsed())),
 		);
 
 		let mut fetched = vec![];

src/network/p2p.rs

@@ -115,7 +115,6 @@ pub fn init(
 	cfg: LibP2PConfig,
 	dht_parallelization_limit: usize,
 	ttl: u64,
-	put_batch_size: usize,
 	is_fat_client: bool,
 	id_keys: libp2p::identity::Keypair,
 ) -> Result<(Client, EventLoop)> {
@@ -206,12 +205,7 @@ pub fn init(
 	let (command_sender, command_receiver) = mpsc::channel(10000);
 
 	Ok((
-		Client::new(
-			command_sender,
-			dht_parallelization_limit,
-			ttl,
-			put_batch_size,
-		),
+		Client::new(command_sender, dht_parallelization_limit, ttl),
 		EventLoop::new(
 			swarm,
 			command_receiver,