refactor channel receiver

common/channel/src/channel.rs

@@ -3,7 +3,6 @@ use crate::metrics::unbounded_channel;
 use metrics::{Counter, CounterUsize};
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::sync::mpsc::error::TryRecvError;
 use tokio::sync::oneshot;
 use tokio::time::timeout;
 
@@ -31,7 +30,7 @@ impl<N, Req, Res> Channel<N, Req, Res> {
                 chan: sender,
                 metrics_timeout,
             },
-            Receiver { chan: receiver },
+            receiver,
         )
     }
 }
@@ -74,19 +73,7 @@ impl<N, Req, Res> Sender<N, Req, Res> {
     }
 }
 
-pub struct Receiver<N, Req, Res> {
-    chan: crate::metrics::Receiver<Message<N, Req, Res>>,
-}
-
-impl<N, Req, Res> Receiver<N, Req, Res> {
-    pub async fn recv(&mut self) -> Option<Message<N, Req, Res>> {
-        self.chan.recv().await
-    }
-
-    pub fn try_recv(&mut self) -> Result<Message<N, Req, Res>, TryRecvError> {
-        self.chan.try_recv()
-    }
-}
+pub type Receiver<N, Req, Res> = crate::metrics::Receiver<Message<N, Req, Res>>;
 
 #[cfg(test)]
 mod tests {

node/sync/src/controllers/metrics.rs

@@ -6,5 +6,6 @@ lazy_static::lazy_static! {
     pub static ref SERIAL_SYNC_FILE_COMPLETED: Arc<dyn Timer> = register_timer("sync_controllers_serial_sync_file_completed");
     pub static ref SERIAL_SYNC_CHUNKS_COMPLETED: Arc<dyn Timer> = register_timer("sync_controllers_serial_sync_chunks_completed");
     pub static ref SERIAL_SYNC_SEGMENT_LATENCY: Arc<dyn Histogram> = Sample::ExpDecay(0.015).register("sync_controllers_serial_sync_segment_latency", 1024);
+    pub static ref SERIAL_SYNC_SEGMENT_TIMEOUT: Arc<dyn Counter<usize>> = CounterUsize::register("sync_controllers_serial_sync_segment_timeout");
     pub static ref SERIAL_SYNC_UNEXPECTED_ERRORS: Arc<dyn Counter<usize>> = CounterUsize::register("sync_controllers_serial_sync_unexpected_errors");
 }

node/sync/src/controllers/serial.rs

@@ -684,6 +684,7 @@ impl SerialSyncController {
                         debug!(%self.tx_seq, "No peer to continue downloading and try to find other peers to download");
                         self.state = SyncState::Idle;
                     } else if since.elapsed() >= self.config.peer_chunks_download_timeout {
+                        metrics::SERIAL_SYNC_SEGMENT_TIMEOUT.inc(1);
                         self.handle_response_failure(peer_id, "RPC timeout");
                     } else {
                         completed = true;