refactor: [torrust#1456] increase ban counters asyncronously

Author: josecelano

packages/udp-tracker-server/src/environment.rs

@@ -82,6 +82,7 @@ impl Environment<Stopped> {
         let udp_server_event_listener_job = Some(crate::statistics::event::listener::run_event_listener(
             self.container.udp_tracker_server_container.event_bus.receiver(),
             &self.container.udp_tracker_server_container.stats_repository,
+            &self.container.udp_tracker_core_container.ban_service,
         ));
 
         // Start the UDP tracker server

packages/udp-tracker-server/src/handlers/mod.rs

@@ -13,7 +13,6 @@ use announce::handle_announce;
 use aquatic_udp_protocol::{Request, Response, TransactionId};
 use bittorrent_tracker_core::MAX_SCRAPE_TORRENTS;
 use bittorrent_udp_tracker_core::container::UdpTrackerCoreContainer;
-use bittorrent_udp_tracker_core::services::announce::UdpAnnounceError;
 use connect::handle_connect;
 use error::handle_error;
 use scrape::handle_scrape;
@@ -84,15 +83,6 @@ pub(crate) async fn handle_packet(
             {
                 Ok((response, req_kid)) => return (response, Some(req_kid)),
                 Err((error, transaction_id, req_kind)) => {
-                    if let Error::UdpAnnounceError {
-                        source: UdpAnnounceError::ConnectionCookieError { .. },
-                    } = error
-                    {
-                        // code-review: should we include `RequestParseError` and `BadRequest`?
-                        let mut ban_service = udp_tracker_core_container.ban_service.write().await;
-                        ban_service.increase_counter(&udp_request.from.ip());
-                    }
-
                     let response = handle_error(
                         Some(req_kind.clone()),
                         udp_request.from,

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

@@ -1,8 +1,12 @@
+use std::sync::Arc;
+
+use bittorrent_udp_tracker_core::services::banning::BanService;
+use tokio::sync::RwLock;
 use torrust_tracker_metrics::label::{LabelSet, LabelValue};
 use torrust_tracker_metrics::{label_name, metric_name};
 use torrust_tracker_primitives::DurationSinceUnixEpoch;
 
-use crate::event::{Event, UdpRequestKind, UdpResponseKind};
+use crate::event::{ErrorKind, Event, UdpRequestKind, UdpResponseKind};
 use crate::statistics::repository::Repository;
 use crate::statistics::{
     UDP_TRACKER_SERVER_ERRORS_TOTAL, UDP_TRACKER_SERVER_PERFORMANCE_AVG_PROCESSING_TIME_NS,
@@ -16,7 +20,12 @@ use crate::statistics::{
 /// This function panics if the client IP version does not match the expected
 /// version.
 #[allow(clippy::too_many_lines)]
-pub async fn handle_event(event: Event, stats_repository: &Repository, now: DurationSinceUnixEpoch) {
+pub async fn handle_event(
+    event: Event,
+    stats_repository: &Repository,
+    ban_service: &Arc<RwLock<BanService>>,
+    now: DurationSinceUnixEpoch,
+) {
     match event {
         Event::UdpRequestAborted { context } => {
             // Global fixed metrics
@@ -232,7 +241,14 @@ pub async fn handle_event(event: Event, stats_repository: &Repository, now: Dura
                 Err(err) => tracing::error!("Failed to increase the counter: {}", err),
             };
         }
-        Event::UdpError { context, kind, error: _ } => {
+        Event::UdpError { context, kind, error } => {
+            // Increase the number of errors
+            // code-review: should we ban IP due to other errors too?
+            if let ErrorKind::ConnectionCookie(_msg) = error {
+                let mut ban_service = ban_service.write().await;
+                ban_service.increase_counter(&context.client_socket_addr().ip());
+            }
+
             // Global fixed metrics
             match context.client_socket_addr().ip() {
                 std::net::IpAddr::V4(_) => {
@@ -267,7 +283,9 @@ pub async fn handle_event(event: Event, stats_repository: &Repository, now: Dura
 #[cfg(test)]
 mod tests {
     use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+    use std::sync::Arc;
 
+    use bittorrent_udp_tracker_core::services::banning::BanService;
     use torrust_tracker_clock::clock::Time;
     use torrust_tracker_primitives::service_binding::{Protocol, ServiceBinding};
 
@@ -279,6 +297,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_number_of_aborted_requests_when_it_receives_a_udp_request_aborted_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestAborted {
@@ -292,6 +311,7 @@ mod tests {
                 ),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -304,6 +324,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_number_of_banned_requests_when_it_receives_a_udp_request_banned_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestBanned {
@@ -317,6 +338,7 @@ mod tests {
                 ),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -329,6 +351,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_number_of_incoming_requests_when_it_receives_a_udp4_incoming_request_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestReceived {
@@ -342,6 +365,7 @@ mod tests {
                 ),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -354,6 +378,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp_abort_counter_when_it_receives_a_udp_abort_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestAborted {
@@ -367,6 +392,7 @@ mod tests {
                 ),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -376,6 +402,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp_ban_counter_when_it_receives_a_udp_banned_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestBanned {
@@ -389,6 +416,7 @@ mod tests {
                 ),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -399,6 +427,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp4_connect_requests_counter_when_it_receives_a_udp4_request_event_of_connect_kind() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestAccepted {
@@ -413,6 +442,7 @@ mod tests {
                 kind: crate::event::UdpRequestKind::Connect,
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -425,6 +455,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp4_announce_requests_counter_when_it_receives_a_udp4_request_event_of_announce_kind() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestAccepted {
@@ -439,6 +470,7 @@ mod tests {
                 kind: crate::event::UdpRequestKind::Announce,
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -451,6 +483,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp4_scrape_requests_counter_when_it_receives_a_udp4_request_event_of_scrape_kind() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestAccepted {
@@ -465,6 +498,7 @@ mod tests {
                 kind: crate::event::UdpRequestKind::Scrape,
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -477,6 +511,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp4_responses_counter_when_it_receives_a_udp4_response_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpResponseSent {
@@ -494,6 +529,7 @@ mod tests {
                 req_processing_time: std::time::Duration::from_secs(1),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -506,6 +542,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp4_errors_counter_when_it_receives_a_udp4_error_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpError {
@@ -521,6 +558,7 @@ mod tests {
                 error: ErrorKind::RequestParse("Invalid request format".to_string()),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -533,6 +571,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp6_connect_requests_counter_when_it_receives_a_udp6_request_event_of_connect_kind() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestAccepted {
@@ -547,6 +586,7 @@ mod tests {
                 kind: crate::event::UdpRequestKind::Connect,
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -559,6 +599,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp6_announce_requests_counter_when_it_receives_a_udp6_request_event_of_announce_kind() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestAccepted {
@@ -573,6 +614,7 @@ mod tests {
                 kind: crate::event::UdpRequestKind::Announce,
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -585,6 +627,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp6_scrape_requests_counter_when_it_receives_a_udp6_request_event_of_scrape_kind() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpRequestAccepted {
@@ -599,6 +642,7 @@ mod tests {
                 kind: crate::event::UdpRequestKind::Scrape,
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -611,6 +655,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp6_response_counter_when_it_receives_a_udp6_response_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpResponseSent {
@@ -628,6 +673,7 @@ mod tests {
                 req_processing_time: std::time::Duration::from_secs(1),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;
@@ -639,6 +685,7 @@ mod tests {
     #[tokio::test]
     async fn should_increase_the_udp6_errors_counter_when_it_receives_a_udp6_error_event() {
         let stats_repository = Repository::new();
+        let ban_service = Arc::new(tokio::sync::RwLock::new(BanService::new(1)));
 
         handle_event(
             Event::UdpError {
@@ -654,6 +701,7 @@ mod tests {
                 error: ErrorKind::RequestParse("Invalid request format".to_string()),
             },
             &stats_repository,
+            &ban_service,
             CurrentClock::now(),
         )
         .await;

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

@@ -1,6 +1,8 @@
 use std::sync::Arc;
 
+use bittorrent_udp_tracker_core::services::banning::BanService;
 use bittorrent_udp_tracker_core::UDP_TRACKER_LOG_TARGET;
+use tokio::sync::RwLock;
 use tokio::task::JoinHandle;
 use torrust_tracker_clock::clock::Time;
 use torrust_tracker_events::receiver::RecvError;
@@ -11,19 +13,24 @@ use crate::statistics::repository::Repository;
 use crate::CurrentClock;
 
 #[must_use]
-pub fn run_event_listener(receiver: Receiver, repository: &Arc<Repository>) -> JoinHandle<()> {
-    let stats_repository = repository.clone();
+pub fn run_event_listener(
+    receiver: Receiver,
+    repository: &Arc<Repository>,
+    ban_service: &Arc<RwLock<BanService>>,
+) -> JoinHandle<()> {
+    let repository_clone = repository.clone();
+    let ban_service_clone = ban_service.clone();
 
     tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Starting UDP tracker server event listener");
 
     tokio::spawn(async move {
-        dispatch_events(receiver, stats_repository).await;
+        dispatch_events(receiver, repository_clone, ban_service_clone).await;
 
         tracing::info!(target: UDP_TRACKER_LOG_TARGET, "UDP tracker server event listener finished");
     })
 }
 
-async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repository>) {
+async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repository>, ban_service: Arc<RwLock<BanService>>) {
     let shutdown_signal = tokio::signal::ctrl_c();
     tokio::pin!(shutdown_signal);
 
@@ -38,7 +45,7 @@ async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repositor
 
             result = receiver.recv() => {
                 match result {
-                    Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
+                    Ok(event) => handle_event(event, &stats_repository, &ban_service, CurrentClock::now()).await,
                     Err(e) => {
                         match e {
                             RecvError::Closed => {

src/bootstrap/jobs/udp_tracker_server.rs

@@ -10,6 +10,7 @@ pub fn start_event_listener(config: &Configuration, app_container: &Arc<AppConta
         let job = torrust_udp_tracker_server::statistics::event::listener::run_event_listener(
             app_container.udp_tracker_server_container.event_bus.receiver(),
             &app_container.udp_tracker_server_container.stats_repository,
+            &app_container.udp_tracker_core_services.ban_service,
         );
         Some(job)
     } else {