fix: [#1449] don't stop stats listeners when lagged

If the stats listener is lagged we continue processing new events even
if metrics become imprecise after it.

This is a temporary solution. See
#1449.

Author: josecelano

packages/http-tracker-core/src/lib.rs

@@ -16,6 +16,8 @@ pub(crate) type CurrentClock = clock::Working;
 #[allow(dead_code)]
 pub(crate) type CurrentClock = clock::Stopped;
 
+pub const HTTP_TRACKER_LOG_TARGET: &str = "HTTP TRACKER";
+
 #[cfg(test)]
 pub(crate) mod tests {
     use bittorrent_primitives::info_hash::InfoHash;

packages/http-tracker-core/src/statistics/event/listener.rs

@@ -4,15 +4,23 @@ use torrust_tracker_clock::clock::Time;
 use super::handler::handle_event;
 use crate::event::Event;
 use crate::statistics::repository::Repository;
-use crate::CurrentClock;
+use crate::{CurrentClock, HTTP_TRACKER_LOG_TARGET};
 
 pub async fn dispatch_events(mut receiver: broadcast::Receiver<Event>, stats_repository: Repository) {
     loop {
         match receiver.recv().await {
             Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
             Err(e) => {
-                tracing::error!("Error receiving http tracker core event: {:?}", e);
-                break;
+                match e {
+                    broadcast::error::RecvError::Closed => {
+                        tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Http core statistics receiver closed.");
+                        break;
+                    }
+                    broadcast::error::RecvError::Lagged(n) => {
+                        // From now on, metrics will be imprecise
+                        tracing::warn!(target: HTTP_TRACKER_LOG_TARGET, "Http core statistics receiver lagged by {} events.", n);
+                    }
+                }
             }
         }
     }

packages/udp-tracker-core/src/statistics/event/listener.rs

@@ -4,15 +4,23 @@ use torrust_tracker_clock::clock::Time;
 use super::handler::handle_event;
 use crate::event::Event;
 use crate::statistics::repository::Repository;
-use crate::CurrentClock;
+use crate::{CurrentClock, UDP_TRACKER_LOG_TARGET};
 
 pub async fn dispatch_events(mut receiver: broadcast::Receiver<Event>, stats_repository: Repository) {
     loop {
         match receiver.recv().await {
             Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
             Err(e) => {
-                tracing::error!("Error receiving udp tracker core event: {:?}", e);
-                break;
+                match e {
+                    broadcast::error::RecvError::Closed => {
+                        tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp core statistics receiver closed.");
+                        break;
+                    }
+                    broadcast::error::RecvError::Lagged(n) => {
+                        // From now on, metrics will be imprecise
+                        tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp core statistics receiver lagged by {} events.", n);
+                    }
+                }
             }
         }
     }

packages/udp-tracker-server/src/statistics/event/listener.rs

@@ -1,3 +1,4 @@
+use bittorrent_udp_tracker_core::UDP_TRACKER_LOG_TARGET;
 use tokio::sync::broadcast;
 use torrust_tracker_clock::clock::Time;
 
@@ -11,8 +12,16 @@ pub async fn dispatch_events(mut receiver: broadcast::Receiver<Event>, stats_rep
         match receiver.recv().await {
             Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
             Err(e) => {
-                tracing::error!("Error receiving udp tracker server event: {:?}", e);
-                break;
+                match e {
+                    broadcast::error::RecvError::Closed => {
+                        tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp server statistics receiver closed.");
+                        break;
+                    }
+                    broadcast::error::RecvError::Lagged(n) => {
+                        // From now on, metrics will be imprecise
+                        tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp server statistics receiver lagged by {} events.", n);
+                    }
+                }
             }
         }
     }

src/bootstrap/jobs/torrent_cleanup.rs

@@ -37,7 +37,7 @@ pub fn start_job(config: &Core, torrents_manager: &Arc<TorrentsManager>) -> Join
         loop {
             tokio::select! {
                 _ = tokio::signal::ctrl_c() => {
-                    tracing::info!("Stopping torrent cleanup job..");
+                    tracing::info!("Stopping torrent cleanup job ...");
                     break;
                 }
                 _ = interval.tick() => {