Merge #1587: Graceful shutdown for event listeners

d2e75e3 refactor: [#1405] gracefull shutdown for listeners (Jose Celano)

Pull request description:

  Events listeners listen for the cancellation request instead of directly for the CRTR+c signal. This will allow implementing centralized policies for shutdown and alternative conditions.

  In the future, we should implement the same for all tasks, including:

  - HTTP servers using axum: `torrust_server_lib::signals::global_shutdown_signal`
  - Job to update metrics: `torrust_tracker_swarm_coordination_registry::statistics::activity_metrics_updater::start_job`
  - Torrent cleanup jonb: `torrust_tracker_lib::bootstrap::jobs::torrent_cleanup::start_job`

ACKs for top commit:
  josecelano:
    ACK d2e75e3

Tree-SHA512: dfdc65b9e16237ac97bbb3229cd9782bbcdc45706731b7d3fec9397770c406d98ec1f16024ad21dee851a55c70451f00c47c08c9f6983364f74a8f533f9e57c6

Author: josecelano

Cargo.lock

@@ -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" }

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/Cargo.toml

@@ -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/environment.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/server.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/announce.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/axum-http-tracker-server/src/v1/handlers/scrape.rs

@@ -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/events/src/shutdown.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);
     }
 
     (