Alert on frequent network errors (#7410)

* Introduce is_frequent util

* Add dirty warn_if_frequent! implementation

* Add freq

* Fix order in condition

* Update

* Update docs

* Fix

* Remove old impl

* Fix errors

* Add wif to av-distr

* Add wif to col prot

* Rename

* Add wif to state-distr

* Address review comments

* Change Freq implementation

* Remove the zero division check

* Make rate explicit

* Fix typo

* Update rate constant

* Introduce explicit rates

* Update docs

* Split errors freq

* Downgrade coarsetime

polkadot/node/network/availability-distribution/src/error.rs

@@ -91,7 +91,11 @@ pub type Result<T> = std::result::Result<T, Error>;
 ///
 /// We basically always want to try and continue on error. This utility function is meant to
 /// consume top-level errors by simply logging them
-pub fn log_error(result: Result<()>, ctx: &'static str) -> std::result::Result<(), FatalError> {
+pub fn log_error(
+	result: Result<()>,
+	ctx: &'static str,
+	warn_freq: &mut gum::Freq,
+) -> std::result::Result<(), FatalError> {
 	match result.into_nested()? {
 		Ok(()) => Ok(()),
 		Err(jfyi) => {
@@ -104,7 +108,8 @@ pub fn log_error(result: Result<()>, ctx: &'static str) -> std::result::Result<(
 				JfyiError::FetchPoV(_) |
 				JfyiError::SendResponse |
 				JfyiError::NoSuchPoV |
-				JfyiError::Runtime(_) => gum::debug!(target: LOG_TARGET, error = ?jfyi, ctx),
+				JfyiError::Runtime(_) =>
+					gum::warn_if_frequent!(freq: warn_freq, max_rate: gum::Times::PerHour(100), target: LOG_TARGET, error = ?jfyi, ctx),
 			}
 			Ok(())
 		},

polkadot/node/network/availability-distribution/src/lib.rs

@@ -97,6 +97,7 @@ impl AvailabilityDistributionSubsystem {
 
 		let IncomingRequestReceivers { pov_req_receiver, chunk_req_receiver } = recvs;
 		let mut requester = Requester::new(metrics.clone()).fuse();
+		let mut warn_freq = gum::Freq::new();
 
 		{
 			let sender = ctx.sender().clone();
@@ -147,6 +148,7 @@ impl AvailabilityDistributionSubsystem {
 							.update_fetching_heads(&mut ctx, &mut runtime, update, &spans)
 							.await,
 						"Error in Requester::update_fetching_heads",
+						&mut warn_freq,
 					)?;
 				},
 				FromOrchestra::Signal(OverseerSignal::BlockFinalized(hash, _)) => {
@@ -188,6 +190,7 @@ impl AvailabilityDistributionSubsystem {
 						)
 						.await,
 						"pov_requester::fetch_pov",
+						&mut warn_freq,
 					)?;
 				},
 			}

polkadot/node/network/availability-distribution/src/requester/fetch_task/mod.rs

@@ -260,6 +260,8 @@ impl RunningTask {
 		let mut succeeded = false;
 		let mut count: u32 = 0;
 		let mut span = self.span.child("run-fetch-chunk-task").with_relay_parent(self.relay_parent);
+		let mut network_error_freq = gum::Freq::new();
+		let mut canceled_freq = gum::Freq::new();
 		// Try validators in reverse order:
 		while let Some(validator) = self.group.pop() {
 			// Report retries:
@@ -272,7 +274,10 @@ impl RunningTask {
 				.with_chunk_index(self.request.index.0)
 				.with_stage(jaeger::Stage::AvailabilityDistribution);
 			// Send request:
-			let resp = match self.do_request(&validator).await {
+			let resp = match self
+				.do_request(&validator, &mut network_error_freq, &mut canceled_freq)
+				.await
+			{
 				Ok(resp) => resp,
 				Err(TaskError::ShuttingDown) => {
 					gum::info!(
@@ -342,6 +347,8 @@ impl RunningTask {
 	async fn do_request(
 		&mut self,
 		validator: &AuthorityDiscoveryId,
+		nerwork_error_freq: &mut gum::Freq,
+		canceled_freq: &mut gum::Freq,
 	) -> std::result::Result<ChunkFetchingResponse, TaskError> {
 		gum::trace!(
 			target: LOG_TARGET,
@@ -386,7 +393,9 @@ impl RunningTask {
 				Err(TaskError::PeerError)
 			},
 			Err(RequestError::NetworkError(err)) => {
-				gum::debug!(
+				gum::warn_if_frequent!(
+					freq: nerwork_error_freq,
+					max_rate: gum::Times::PerHour(100),
 					target: LOG_TARGET,
 					origin = ?validator,
 					relay_parent = ?self.relay_parent,
@@ -400,7 +409,9 @@ impl RunningTask {
 				Err(TaskError::PeerError)
 			},
 			Err(RequestError::Canceled(oneshot::Canceled)) => {
-				gum::debug!(
+				gum::warn_if_frequent!(
+					freq: canceled_freq,
+					max_rate: gum::Times::PerHour(100),
 					target: LOG_TARGET,
 					origin = ?validator,
 					relay_parent = ?self.relay_parent,

polkadot/node/network/collator-protocol/src/validator_side/mod.rs

@@ -1270,6 +1270,9 @@ async fn run_inner<Context>(
 	let check_collations_stream = tick_stream(CHECK_COLLATIONS_POLL);
 	futures::pin_mut!(check_collations_stream);
 
+	let mut network_error_freq = gum::Freq::new();
+	let mut canceled_freq = gum::Freq::new();
+
 	loop {
 		select! {
 			_ = reputation_delay => {
@@ -1312,6 +1315,8 @@ async fn run_inner<Context>(
 					&mut state.requested_collations,
 					&state.metrics,
 					&state.span_per_relay_parent,
+					&mut network_error_freq,
+					&mut canceled_freq,
 				).await;
 
 				for (peer_id, rep) in reputation_changes {
@@ -1328,14 +1333,22 @@ async fn poll_requests(
 	requested_collations: &mut HashMap<PendingCollation, PerRequest>,
 	metrics: &Metrics,
 	span_per_relay_parent: &HashMap<Hash, PerLeafSpan>,
+	network_error_freq: &mut gum::Freq,
+	canceled_freq: &mut gum::Freq,
 ) -> Vec<(PeerId, Rep)> {
 	let mut retained_requested = HashSet::new();
 	let mut reputation_changes = Vec::new();
 	for (pending_collation, per_req) in requested_collations.iter_mut() {
 		// Despite the await, this won't block on the response itself.
-		let result =
-			poll_collation_response(metrics, span_per_relay_parent, pending_collation, per_req)
-				.await;
+		let result = poll_collation_response(
+			metrics,
+			span_per_relay_parent,
+			pending_collation,
+			per_req,
+			network_error_freq,
+			canceled_freq,
+		)
+		.await;
 
 		if !result.is_ready() {
 			retained_requested.insert(pending_collation.clone());
@@ -1479,6 +1492,8 @@ async fn poll_collation_response(
 	spans: &HashMap<Hash, PerLeafSpan>,
 	pending_collation: &PendingCollation,
 	per_req: &mut PerRequest,
+	network_error_freq: &mut gum::Freq,
+	canceled_freq: &mut gum::Freq,
 ) -> CollationFetchResult {
 	if never!(per_req.from_collator.is_terminated()) {
 		gum::error!(
@@ -1522,7 +1537,9 @@ async fn poll_collation_response(
 				CollationFetchResult::Error(None)
 			},
 			Err(RequestError::NetworkError(err)) => {
-				gum::debug!(
+				gum::warn_if_frequent!(
+					freq: network_error_freq,
+					max_rate: gum::Times::PerHour(100),
 					target: LOG_TARGET,
 					hash = ?pending_collation.relay_parent,
 					para_id = ?pending_collation.para_id,
@@ -1537,7 +1554,9 @@ async fn poll_collation_response(
 				CollationFetchResult::Error(Some(COST_NETWORK_ERROR))
 			},
 			Err(RequestError::Canceled(err)) => {
-				gum::debug!(
+				gum::warn_if_frequent!(
+					freq: canceled_freq,
+					max_rate: gum::Times::PerHour(100),
 					target: LOG_TARGET,
 					hash = ?pending_collation.relay_parent,
 					para_id = ?pending_collation.para_id,

polkadot/node/network/statement-distribution/src/error.rs

@@ -82,13 +82,18 @@ pub enum Error {
 ///
 /// We basically always want to try and continue on error. This utility function is meant to
 /// consume top-level errors by simply logging them.
-pub fn log_error(result: Result<()>, ctx: &'static str) -> std::result::Result<(), FatalError> {
+pub fn log_error(
+	result: Result<()>,
+	ctx: &'static str,
+	warn_freq: &mut gum::Freq,
+) -> std::result::Result<(), FatalError> {
 	match result.into_nested()? {
 		Err(jfyi) => {
 			match jfyi {
 				JfyiError::RequestedUnannouncedCandidate(_, _) =>
 					gum::warn!(target: LOG_TARGET, error = %jfyi, ctx),
-				_ => gum::debug!(target: LOG_TARGET, error = %jfyi, ctx),
+				_ =>
+					gum::warn_if_frequent!(freq: warn_freq, max_rate: gum::Times::PerHour(100), target: LOG_TARGET, error = %jfyi, ctx),
 			}
 			Ok(())
 		},

polkadot/node/network/statement-distribution/src/lib.rs

@@ -1810,6 +1810,8 @@ impl<R: rand::Rng> StatementDistributionSubsystem<R> {
 		)
 		.map_err(FatalError::SpawnTask)?;
 
+		let mut warn_freq = gum::Freq::new();
+
 		loop {
 			select! {
 				_ = reputation_delay => {
@@ -1851,7 +1853,7 @@ impl<R: rand::Rng> StatementDistributionSubsystem<R> {
 									result.ok_or(FatalError::RequesterReceiverFinished)?,
 								)
 								.await;
-							log_error(result.map_err(From::from), "handle_requester_message")?;
+							log_error(result.map_err(From::from), "handle_requester_message", &mut warn_freq)?;
 						},
 						MuxedMessage::Responder(result) => {
 							let result = self
@@ -1861,7 +1863,7 @@ impl<R: rand::Rng> StatementDistributionSubsystem<R> {
 									result.ok_or(FatalError::ResponderReceiverFinished)?,
 								)
 								.await;
-							log_error(result.map_err(From::from), "handle_responder_message")?;
+							log_error(result.map_err(From::from), "handle_responder_message", &mut warn_freq)?;
 						},
 					};
 				}