Removed Metrics from the Event Loop

Cargo.toml

@@ -36,6 +36,7 @@ multihash = { version = "0.14.0", default-features = false, features = ["blake3"
 semver = "1.0.23"
 serde = { version = "1.0.163", features = ["derive"] }
 tokio = { version = "1.35", features = ["full"] }
+tokio-stream = { version = "0.1.15", features = ["sync"] }
 tracing = "0.1.35"
 tracing-subscriber = { version = "0.3.17", features = ["json", "env-filter"] }
 uuid = { version = "1.3.4", features = ["v4", "fast-rng", "macro-diagnostics", "serde"] }

client/Cargo.toml

@@ -19,6 +19,7 @@ libp2p = { workspace = true }
 sp-core = { version = "28.0.0", features = ["serde"] }
 strip-ansi-escapes = "0.2.0"
 tokio = { workspace = true }
+tokio-stream = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 uuid = { workspace = true }

client/src/main.rs

@@ -29,8 +29,7 @@ use color_eyre::{
 use kate_recovery::com::AppData;
 use std::{fs, path::Path, sync::Arc};
 use tokio::sync::broadcast;
-use tracing::trace;
-use tracing::{error, info, span, warn, Level};
+use tracing::{error, info, span, trace, warn, Level};
 
 #[cfg(feature = "network-analysis")]
 use avail_light_core::network::p2p::analyzer;
@@ -86,7 +85,7 @@ async fn run(
 		.wrap_err("Unable to initialize OpenTelemetry service")?,
 	);
 
-	let (p2p_client, p2p_event_loop, _) = p2p::init(
+	let (p2p_client, p2p_event_loop, event_receiver) = p2p::init(
 		cfg.libp2p.clone(),
 		id_keys,
 		version,
@@ -98,7 +97,15 @@ async fn run(
 	)
 	.await?;
 
-	spawn_in_span(shutdown.with_cancel(p2p_event_loop.run(ot_metrics.clone())));
+	let metrics_clone = ot_metrics.clone();
+	let shutdown_clone = shutdown.clone();
+	tokio::spawn(async move {
+		shutdown_clone
+			.with_cancel(metrics_clone.handle_event_stream(event_receiver))
+			.await
+	});
+
+	spawn_in_span(shutdown.with_cancel(p2p_event_loop.run()));
 
 	let addrs = vec![
 		cfg.libp2p.tcp_multiaddress(),
@@ -109,7 +116,7 @@ async fn run(
 	p2p_client
 		.start_listening(addrs)
 		.await
-		.wrap_err("Error starting li.")?;
+		.wrap_err("Error starting listener.")?;
 	info!(
 		"TCP listener started on port {}. WebRTC listening on port {}.",
 		cfg.libp2p.port, cfg.libp2p.webrtc_port

core/Cargo.toml

@@ -53,7 +53,7 @@ sysinfo = "0.30.12"
 threadpool = "1.8.1"
 tokio = { workspace = true }
 tokio-retry = "0.3"
-tokio-stream = { version = "0.1.14", features = ["sync"] }
+tokio-stream = { workspace = true }
 tokio-util = "0.7.10"
 tracing = { workspace = true }
 uuid = { workspace = true }

core/src/maintenance.rs

@@ -51,14 +51,13 @@ pub async fn process_block(
 
 	// Reconfigure Kademlia mode if needed
 	if maintenance_config.automatic_server_mode {
-		let new_mode = p2p_client
+		p2p_client
 			.reconfigure_kademlia_mode(
 				maintenance_config.total_memory_gb_threshold,
 				maintenance_config.num_cpus_threshold,
 			)
 			.await
 			.wrap_err("Unable to reconfigure kademlia mode")?;
-		metrics.update_operating_mode(new_mode).await;
 	}
 
 	let peers_num_metric = MetricValue::DHTConnectedPeers(peers_num);

core/src/network/p2p.rs

@@ -4,7 +4,7 @@ use configuration::LibP2PConfig;
 use libp2p::{
 	autonat, dcutr, identify,
 	identity::{self, ed25519, Keypair},
-	kad::{self, Mode, PeerRecord, QueryStats},
+	kad::{self, Mode, PeerRecord},
 	mdns, noise, ping, relay,
 	swarm::NetworkBehaviour,
 	tcp, upnp, yamux, Multiaddr, PeerId, Swarm, SwarmBuilder,
@@ -66,19 +66,7 @@ pub enum OutputEvent {
 	IncomingGetRecord,
 	IncomingPutRecord,
 	KadModeChange(Mode),
-	PutRecordOk {
-		key: kad::RecordKey,
-		stats: QueryStats,
-	},
-	PutRecordQuorumFailed {
-		key: kad::RecordKey,
-		stats: QueryStats,
-	},
-	PutRecordTimeout {
-		key: kad::RecordKey,
-		stats: QueryStats,
-	},
-	NatStatusPrivate,
+	PutRecord { success_rate: f64, duration: f64 },
 	Ping(Duration),
 	IncomingConnection,
 	IncomingConnectionError,

core/src/network/p2p/client.rs

@@ -20,8 +20,8 @@ use tokio::sync::{mpsc::UnboundedSender, oneshot};
 use tracing::{debug, info, trace};
 
 use super::{
-	event_loop::ConnectionEstablishedInfo, is_global, is_multiaddr_global, Command, EventLoop,
-	MultiAddressInfo, PeerInfo, QueryChannel,
+	event_loop::{BlockStat, ConnectionEstablishedInfo},
+	is_global, is_multiaddr_global, Command, EventLoop, MultiAddressInfo, PeerInfo, QueryChannel,
 };
 use crate::types::MultiaddrConfig;
 
@@ -67,22 +67,6 @@ impl DHTRow {
 	}
 }
 
-#[derive(Debug)]
-pub struct BlockStat {
-	pub total_count: usize,
-	pub remaining_counter: usize,
-	pub success_counter: usize,
-	pub error_counter: usize,
-	pub time_stat: u64,
-}
-
-impl BlockStat {
-	pub fn increase_block_stat_counters(&mut self, cell_number: usize) {
-		self.total_count += cell_number;
-		self.remaining_counter += cell_number;
-	}
-}
-
 impl Client {
 	pub fn new(
 		sender: UnboundedSender<Command>,
@@ -337,6 +321,12 @@ impl Client {
 						"Encountered error while sending Reconfigure Kademlia Mode response: {e:?}"
 					)
 				})?;
+
+				context
+					.event_sender
+					.send(super::OutputEvent::KadModeChange(context.kad_mode))
+					.map_err(|e| eyre!("Error while sending Kad Mode Output Event: {e}"))?;
+
 				Ok(())
 			})
 		})