Merge #1451: Temporary solution to handle lagged stats listeners

6fdbc47 fix: [#1449] don't stop stats listeners when lagged (Jose Celano)
ed8acac fix: [#1449] increase broadcast channel capacity (Jose Celano)

Pull request description:

  Relates to: #1449

  When I switched the channels for statistics from [tokio mpsc](https://docs.rs/tokio/latest/tokio/sync/mpsc/index.html) to [tokio broadcast](https://docs.rs/tokio/latest/tokio/sync/broadcast/index.html), I introduced a bug that was only revealed after deploying the demo tracker.

  The problem is that MPSC provides backpressure, so all events are processed. However, the broadcast channel has a limit, and listeners are informed with a `Lagged` error when the limit has been reached and events have been deleted.

  We need to decide if:

  1. We go back to mpsc channel
  2. Or we implement another solution
  3. Or we accept imprecise metrics on a high load

  In the meantime, I have increased the channel capacity to be able to collect all events at the current demo peak load. That should avoid the problem of loosing events in practice.

  I have also fixed the listener to handle the `Lagged` error. In this case, we don't have to break the loop but continue processing events to update metrics. The only difference is metrics will not be precise after this error happens because some events were lost.

ACKs for top commit:
  josecelano:
    ACK 6fdbc47

Tree-SHA512: 3bc60ddb253cf8b6873abd5e6300f8731ee677338c6944360caa10a023d470c258ff163f48d5d4e1c0350266a73e6709910a30e728890ff5295353c262b46380

Author: josecelano

packages/http-tracker-core/src/event/sender.rs

@@ -7,7 +7,7 @@ use tokio::sync::broadcast::error::SendError;
 
 use super::Event;
 
-const CHANNEL_CAPACITY: usize = 1024;
+const CHANNEL_CAPACITY: usize = 32768;
 
 /// A trait for sending sending.
 #[cfg_attr(test, automock)]

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/event/sender.rs

@@ -7,7 +7,7 @@ use tokio::sync::broadcast::error::SendError;
 
 use super::Event;
 
-const CHANNEL_CAPACITY: usize = 1024;
+const CHANNEL_CAPACITY: usize = 32768;
 
 /// A trait for sending sending.
 #[cfg_attr(test, automock)]

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/event/sender.rs

@@ -7,7 +7,7 @@ use tokio::sync::broadcast::error::SendError;
 
 use super::Event;
 
-const CHANNEL_CAPACITY: usize = 1024;
+const CHANNEL_CAPACITY: usize = 32768;
 
 /// A trait for sending sending.
 #[cfg_attr(test, automock)]

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() => {