Graceful shutdown for event listeners

Cargo.toml

@@ -47,6 +47,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = { version = "1", features = ["preserve_order"] }
 thiserror = "2.0.12"
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
+tokio-util = "0.7.15"
 torrust-axum-health-check-api-server = { version = "3.0.0-develop", path = "packages/axum-health-check-api-server" }
 torrust-axum-http-tracker-server = { version = "3.0.0-develop", path = "packages/axum-http-tracker-server" }
 torrust-axum-rest-tracker-api-server = { version = "3.0.0-develop", path = "packages/axum-rest-tracker-api-server" }

packages/axum-http-tracker-server/Cargo.toml

@@ -28,6 +28,7 @@ hyper = "1"
 reqwest = { version = "0", features = ["json"] }
 serde = { version = "1", features = ["derive"] }
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
+tokio-util = "0.7.15"
 torrust-axum-server = { version = "3.0.0-develop", path = "../axum-server" }
 torrust-server-lib = { version = "3.0.0-develop", path = "../server-lib" }
 torrust-tracker-clock = { version = "3.0.0-develop", path = "../clock" }

packages/axum-http-tracker-server/src/environment.rs

@@ -6,6 +6,7 @@ use bittorrent_primitives::info_hash::InfoHash;
 use bittorrent_tracker_core::container::TrackerCoreContainer;
 use futures::executor::block_on;
 use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
 use torrust_axum_server::tsl::make_rust_tls;
 use torrust_server_lib::registar::Registar;
 use torrust_tracker_configuration::{logging, Configuration};
@@ -21,6 +22,7 @@ pub struct Environment<S> {
     pub registar: Registar,
     pub server: HttpServer<S>,
     pub event_listener_job: Option<JoinHandle<()>>,
+    pub cancellation_token: CancellationToken,
 }
 
 impl<S> Environment<S> {
@@ -59,6 +61,7 @@ impl Environment<Stopped> {
             registar: Registar::default(),
             server,
             event_listener_job: None,
+            cancellation_token: CancellationToken::new(),
         }
     }
 
@@ -72,6 +75,7 @@ impl Environment<Stopped> {
         // Start the event listener
         let event_listener_job = run_event_listener(
             self.container.http_tracker_core_container.event_bus.receiver(),
+            self.cancellation_token.clone(),
             &self.container.http_tracker_core_container.stats_repository,
         );
 
@@ -87,6 +91,7 @@ impl Environment<Stopped> {
             registar: self.registar.clone(),
             server,
             event_listener_job: Some(event_listener_job),
+            cancellation_token: self.cancellation_token,
         }
     }
 }
@@ -117,6 +122,7 @@ impl Environment<Running> {
             registar: Registar::default(),
             server,
             event_listener_job: None,
+            cancellation_token: self.cancellation_token,
         }
     }
 

packages/axum-http-tracker-server/src/server.rs

@@ -256,6 +256,7 @@ mod tests {
     use bittorrent_http_tracker_core::statistics::event::listener::run_event_listener;
     use bittorrent_http_tracker_core::statistics::repository::Repository;
     use bittorrent_tracker_core::container::TrackerCoreContainer;
+    use tokio_util::sync::CancellationToken;
     use torrust_axum_server::tsl::make_rust_tls;
     use torrust_server_lib::registar::Registar;
     use torrust_tracker_configuration::{logging, Configuration};
@@ -265,6 +266,8 @@ mod tests {
     use crate::server::{HttpServer, Launcher};
 
     pub fn initialize_container(configuration: &Configuration) -> HttpTrackerCoreContainer {
+        let cancellation_token = CancellationToken::new();
+
         let core_config = Arc::new(configuration.core.clone());
 
         let http_trackers = configuration
@@ -287,7 +290,7 @@ mod tests {
         let http_stats_event_sender = http_stats_event_bus.sender();
 
         if configuration.core.tracker_usage_statistics {
-            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), cancellation_token, &http_stats_repository);
         }
 
         let swarm_coordination_registry_container = Arc::new(SwarmCoordinationRegistryContainer::initialize(

packages/axum-http-tracker-server/src/v1/handlers/announce.rs

@@ -123,6 +123,7 @@ mod tests {
     use bittorrent_tracker_core::torrent::repository::in_memory::InMemoryTorrentRepository;
     use bittorrent_tracker_core::whitelist::authorization::WhitelistAuthorization;
     use bittorrent_tracker_core::whitelist::repository::in_memory::InMemoryWhitelist;
+    use tokio_util::sync::CancellationToken;
     use torrust_tracker_configuration::Configuration;
     use torrust_tracker_test_helpers::configuration;
 
@@ -149,6 +150,9 @@ mod tests {
     }
 
     fn initialize_core_tracker_services(config: &Configuration) -> CoreHttpTrackerServices {
+        let cancellation_token = CancellationToken::new();
+
+        // Initialize the core tracker services with the provided configuration.
         let core_config = Arc::new(config.core.clone());
         let database = initialize_database(&config.core);
         let in_memory_whitelist = Arc::new(InMemoryWhitelist::default());
@@ -175,7 +179,7 @@ mod tests {
         let http_stats_event_sender = http_stats_event_bus.sender();
 
         if config.core.tracker_usage_statistics {
-            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), cancellation_token, &http_stats_repository);
         }
 
         let announce_service = Arc::new(AnnounceService::new(

packages/axum-http-tracker-server/src/v1/handlers/scrape.rs

@@ -97,6 +97,7 @@ mod tests {
     use bittorrent_tracker_core::torrent::repository::in_memory::InMemoryTorrentRepository;
     use bittorrent_tracker_core::whitelist::authorization::WhitelistAuthorization;
     use bittorrent_tracker_core::whitelist::repository::in_memory::InMemoryWhitelist;
+    use tokio_util::sync::CancellationToken;
     use torrust_tracker_configuration::{Configuration, Core};
     use torrust_tracker_test_helpers::configuration;
 
@@ -127,6 +128,8 @@ mod tests {
     }
 
     fn initialize_core_tracker_services(config: &Configuration) -> (CoreTrackerServices, CoreHttpTrackerServices) {
+        let cancellation_token = CancellationToken::new();
+
         let core_config = Arc::new(config.core.clone());
         let in_memory_whitelist = Arc::new(InMemoryWhitelist::default());
         let whitelist_authorization = Arc::new(WhitelistAuthorization::new(&config.core, &in_memory_whitelist.clone()));
@@ -146,7 +149,7 @@ mod tests {
         let http_stats_event_sender = http_stats_event_bus.sender();
 
         if config.core.tracker_usage_statistics {
-            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), cancellation_token, &http_stats_repository);
         }
 
         (

packages/http-tracker-core/Cargo.toml

@@ -23,6 +23,7 @@ futures = "0"
 serde = "1.0.219"
 thiserror = "2"
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
+tokio-util = "0.7.15"
 torrust-tracker-clock = { version = "3.0.0-develop", path = "../clock" }
 torrust-tracker-configuration = { version = "3.0.0-develop", path = "../configuration" }
 torrust-tracker-events = { version = "3.0.0-develop", path = "../events" }

packages/http-tracker-core/benches/helpers/util.rs

@@ -20,6 +20,7 @@ use bittorrent_tracker_core::whitelist::authorization::WhitelistAuthorization;
 use bittorrent_tracker_core::whitelist::repository::in_memory::InMemoryWhitelist;
 use futures::future::BoxFuture;
 use mockall::mock;
+use tokio_util::sync::CancellationToken;
 use torrust_tracker_configuration::{Configuration, Core};
 use torrust_tracker_events::sender::SendError;
 use torrust_tracker_primitives::peer::Peer;
@@ -42,6 +43,8 @@ pub fn initialize_core_tracker_services() -> (CoreTrackerServices, CoreHttpTrack
 }
 
 pub fn initialize_core_tracker_services_with_config(config: &Configuration) -> (CoreTrackerServices, CoreHttpTrackerServices) {
+    let cancellation_token = CancellationToken::new();
+
     let core_config = Arc::new(config.core.clone());
     let database = initialize_database(&config.core);
     let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default());
@@ -69,7 +72,7 @@ pub fn initialize_core_tracker_services_with_config(config: &Configuration) -> (
     let http_stats_event_sender = http_stats_event_bus.sender();
 
     if config.core.tracker_usage_statistics {
-        let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
+        let _unused = run_event_listener(http_stats_event_bus.receiver(), cancellation_token, &http_stats_repository);
     }
 
     (

packages/http-tracker-core/src/services/announce.rs

@@ -216,6 +216,7 @@ mod tests {
     use bittorrent_tracker_core::torrent::repository::in_memory::InMemoryTorrentRepository;
     use bittorrent_tracker_core::whitelist::authorization::WhitelistAuthorization;
     use bittorrent_tracker_core::whitelist::repository::in_memory::InMemoryWhitelist;
+    use tokio_util::sync::CancellationToken;
     use torrust_tracker_configuration::{Configuration, Core};
     use torrust_tracker_primitives::peer::Peer;
     use torrust_tracker_test_helpers::configuration;
@@ -236,6 +237,8 @@ mod tests {
     }
 
     fn initialize_core_tracker_services_with_config(config: &Configuration) -> (CoreTrackerServices, CoreHttpTrackerServices) {
+        let cancellation_token = CancellationToken::new();
+
         let core_config = Arc::new(config.core.clone());
         let database = initialize_database(&config.core);
         let in_memory_torrent_repository = Arc::new(InMemoryTorrentRepository::default());
@@ -263,7 +266,7 @@ mod tests {
         let http_stats_event_sender = http_stats_event_bus.sender();
 
         if config.core.tracker_usage_statistics {
-            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), cancellation_token, &http_stats_repository);
         }
 
         (

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

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
 use torrust_tracker_clock::clock::Time;
 use torrust_tracker_events::receiver::RecvError;
 
@@ -10,29 +11,29 @@ use crate::statistics::repository::Repository;
 use crate::{CurrentClock, HTTP_TRACKER_LOG_TARGET};
 
 #[must_use]
-pub fn run_event_listener(receiver: Receiver, repository: &Arc<Repository>) -> JoinHandle<()> {
+pub fn run_event_listener(
+    receiver: Receiver,
+    cancellation_token: CancellationToken,
+    repository: &Arc<Repository>,
+) -> JoinHandle<()> {
     let stats_repository = repository.clone();
 
     tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Starting HTTP tracker core event listener");
 
     tokio::spawn(async move {
-        dispatch_events(receiver, stats_repository).await;
+        dispatch_events(receiver, cancellation_token, stats_repository).await;
 
         tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "HTTP tracker core event listener finished");
     })
 }
 
-async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repository>) {
-    let shutdown_signal = tokio::signal::ctrl_c();
-
-    tokio::pin!(shutdown_signal);
-
+async fn dispatch_events(mut receiver: Receiver, cancellation_token: CancellationToken, stats_repository: Arc<Repository>) {
     loop {
         tokio::select! {
             biased;
 
-            _ = &mut shutdown_signal => {
-                tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Received Ctrl+C, shutting down HTTP tracker core event listener.");
+            () = cancellation_token.cancelled() => {
+                tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Received cancellation request, shutting down HTTP tracker core event listener.");
                 break;
             }
 
@@ -42,11 +43,11 @@ async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repositor
                     Err(e) => {
                         match e {
                             RecvError::Closed => {
-                                tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Http core statistics receiver closed.");
+                                tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Http tracker core statistics receiver closed.");
                                 break;
                             }
                             RecvError::Lagged(n) => {
-                                tracing::warn!(target: HTTP_TRACKER_LOG_TARGET, "Http core statistics receiver lagged by {} events.", n);
+                                tracing::warn!(target: HTTP_TRACKER_LOG_TARGET, "Http tracker core statistics receiver lagged by {} events.", n);
                             }
                         }
                     }

packages/rest-tracker-api-core/Cargo.toml

@@ -18,6 +18,7 @@ bittorrent-http-tracker-core = { version = "3.0.0-develop", path = "../http-trac
 bittorrent-tracker-core = { version = "3.0.0-develop", path = "../tracker-core" }
 bittorrent-udp-tracker-core = { version = "3.0.0-develop", path = "../udp-tracker-core" }
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
+tokio-util = "0.7.15"
 torrust-tracker-configuration = { version = "3.0.0-develop", path = "../configuration" }
 torrust-tracker-metrics = { version = "3.0.0-develop", path = "../metrics" }
 torrust-tracker-primitives = { version = "3.0.0-develop", path = "../primitives" }

packages/rest-tracker-api-core/src/statistics/services.rs

@@ -210,6 +210,7 @@ mod tests {
     use bittorrent_udp_tracker_core::services::banning::BanService;
     use bittorrent_udp_tracker_core::MAX_CONNECTION_ID_ERRORS_PER_IP;
     use tokio::sync::RwLock;
+    use tokio_util::sync::CancellationToken;
     use torrust_tracker_configuration::Configuration;
     use torrust_tracker_events::bus::SenderStatus;
     use torrust_tracker_swarm_coordination_registry::container::SwarmCoordinationRegistryContainer;
@@ -224,6 +225,8 @@ mod tests {
 
     #[tokio::test]
     async fn the_statistics_service_should_return_the_tracker_metrics() {
+        let cancellation_token = CancellationToken::new();
+
         let config = tracker_configuration();
         let core_config = Arc::new(config.core.clone());
 
@@ -244,7 +247,7 @@ mod tests {
         ));
 
         if config.core.tracker_usage_statistics {
-            let _unused = run_event_listener(http_stats_event_bus.receiver(), &http_stats_repository);
+            let _unused = run_event_listener(http_stats_event_bus.receiver(), cancellation_token, &http_stats_repository);
         }
 
         // UDP server stats

packages/swarm-coordination-registry/Cargo.toml

@@ -24,6 +24,7 @@ futures = "0"
 serde = { version = "1.0.219", features = ["derive"] }
 thiserror = "2.0.12"
 tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] }
+tokio-util = "0.7.15"
 torrust-tracker-clock = { version = "3.0.0-develop", path = "../clock" }
 torrust-tracker-configuration = { version = "3.0.0-develop", path = "../configuration" }
 torrust-tracker-events = { version = "3.0.0-develop", path = "../events" }