From d2e75e3f78f367e5f2829bbe5adfedc549fb24f5 Mon Sep 17 00:00:00 2001 From: Jose Celano Date: Tue, 17 Jun 2025 10:57:11 +0100 Subject: [PATCH] refactor: [#1405] gracefull shutdown for listeners Events listeners listen for the cancelation request instead of directly for the CRTR+c signal. This will allow implementing centralized policies for shutdown and alternative conditions. --- Cargo.lock | 8 +++++ Cargo.toml | 1 + packages/axum-http-tracker-server/Cargo.toml | 1 + .../src/environment.rs | 6 ++++ .../axum-http-tracker-server/src/server.rs | 5 +++- .../src/v1/handlers/announce.rs | 6 +++- .../src/v1/handlers/scrape.rs | 5 +++- packages/events/src/shutdown.rs | 0 packages/http-tracker-core/Cargo.toml | 1 + .../http-tracker-core/benches/helpers/util.rs | 5 +++- .../src/services/announce.rs | 5 +++- .../src/statistics/event/listener.rs | 23 +++++++------- packages/rest-tracker-api-core/Cargo.toml | 1 + .../src/statistics/services.rs | 5 +++- .../swarm-coordination-registry/Cargo.toml | 1 + .../src/statistics/event/listener.rs | 27 +++++++++-------- packages/tracker-core/Cargo.toml | 1 + .../src/statistics/event/listener.rs | 20 ++++++------- .../tracker-core/tests/common/test_env.rs | 5 ++++ packages/udp-tracker-core/Cargo.toml | 1 + .../src/statistics/event/listener.rs | 22 +++++++------- packages/udp-tracker-server/Cargo.toml | 1 + .../src/banning/event/listener.rs | 22 ++++++++------ .../udp-tracker-server/src/environment.rs | 9 ++++++ .../src/statistics/event/listener.rs | 22 +++++++------- src/app.rs | 12 ++++---- src/bootstrap/jobs/http_tracker_core.rs | 8 ++++- src/bootstrap/jobs/manager.rs | 30 +++++++++++++++++-- src/bootstrap/jobs/torrent_repository.rs | 8 ++++- src/bootstrap/jobs/tracker_core.rs | 8 ++++- src/bootstrap/jobs/udp_tracker_core.rs | 8 ++++- src/bootstrap/jobs/udp_tracker_server.rs | 11 +++++-- src/main.rs | 2 ++ 33 files changed, 206 insertions(+), 84 deletions(-) create mode 100644 packages/events/src/shutdown.rs diff --git a/Cargo.lock b/Cargo.lock index 269f7a3a2..b523c8b60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -587,6 +587,7 @@ dependencies = [ "serde_json", "thiserror 2.0.12", "tokio", + "tokio-util", "torrust-tracker-clock", "torrust-tracker-configuration", "torrust-tracker-events", @@ -673,6 +674,7 @@ dependencies = [ "testcontainers", "thiserror 2.0.12", "tokio", + "tokio-util", "torrust-rest-tracker-api-client", "torrust-tracker-clock", "torrust-tracker-configuration", @@ -705,6 +707,7 @@ dependencies = [ "serde", "thiserror 2.0.12", "tokio", + "tokio-util", "torrust-tracker-clock", "torrust-tracker-configuration", "torrust-tracker-events", @@ -4565,6 +4568,7 @@ dependencies = [ "serde_bytes", "serde_repr", "tokio", + "tokio-util", "torrust-axum-server", "torrust-server-lib", "torrust-tracker-clock", @@ -4661,6 +4665,7 @@ dependencies = [ "bittorrent-tracker-core", "bittorrent-udp-tracker-core", "tokio", + "tokio-util", "torrust-tracker-configuration", "torrust-tracker-events", "torrust-tracker-metrics", @@ -4704,6 +4709,7 @@ dependencies = [ "serde_json", "thiserror 2.0.12", "tokio", + "tokio-util", "torrust-axum-health-check-api-server", "torrust-axum-http-tracker-server", "torrust-axum-rest-tracker-api-server", @@ -4851,6 +4857,7 @@ dependencies = [ "serde", "thiserror 2.0.12", "tokio", + "tokio-util", "torrust-tracker-clock", "torrust-tracker-configuration", "torrust-tracker-events", @@ -4909,6 +4916,7 @@ dependencies = [ "serde", "thiserror 2.0.12", "tokio", + "tokio-util", "torrust-server-lib", "torrust-tracker-clock", "torrust-tracker-configuration", diff --git a/Cargo.toml b/Cargo.toml index 976176155..dbc39bdf8 100644 --- a/Cargo.toml +++ b/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" } diff --git a/packages/axum-http-tracker-server/Cargo.toml b/packages/axum-http-tracker-server/Cargo.toml index fa195489c..eb2c2cad3 100644 --- a/packages/axum-http-tracker-server/Cargo.toml +++ b/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" } diff --git a/packages/axum-http-tracker-server/src/environment.rs b/packages/axum-http-tracker-server/src/environment.rs index 6e58c2cac..616973a0f 100644 --- a/packages/axum-http-tracker-server/src/environment.rs +++ b/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 { pub registar: Registar, pub server: HttpServer, pub event_listener_job: Option>, + pub cancellation_token: CancellationToken, } impl Environment { @@ -59,6 +61,7 @@ impl Environment { registar: Registar::default(), server, event_listener_job: None, + cancellation_token: CancellationToken::new(), } } @@ -72,6 +75,7 @@ impl Environment { // 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 { registar: self.registar.clone(), server, event_listener_job: Some(event_listener_job), + cancellation_token: self.cancellation_token, } } } @@ -117,6 +122,7 @@ impl Environment { registar: Registar::default(), server, event_listener_job: None, + cancellation_token: self.cancellation_token, } } diff --git a/packages/axum-http-tracker-server/src/server.rs b/packages/axum-http-tracker-server/src/server.rs index ba0dd8c6e..2b43be0a9 100644 --- a/packages/axum-http-tracker-server/src/server.rs +++ b/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( diff --git a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs b/packages/axum-http-tracker-server/src/v1/handlers/announce.rs index e21a485cf..ce718cd30 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/announce.rs +++ b/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( diff --git a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs b/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs index b48d6e036..bdd4378f3 100644 --- a/packages/axum-http-tracker-server/src/v1/handlers/scrape.rs +++ b/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); } ( diff --git a/packages/events/src/shutdown.rs b/packages/events/src/shutdown.rs new file mode 100644 index 000000000..e69de29bb diff --git a/packages/http-tracker-core/Cargo.toml b/packages/http-tracker-core/Cargo.toml index 45af59baa..04a6c96b6 100644 --- a/packages/http-tracker-core/Cargo.toml +++ b/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" } diff --git a/packages/http-tracker-core/benches/helpers/util.rs b/packages/http-tracker-core/benches/helpers/util.rs index 414d3b40e..028d7c535 100644 --- a/packages/http-tracker-core/benches/helpers/util.rs +++ b/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); } ( diff --git a/packages/http-tracker-core/src/services/announce.rs b/packages/http-tracker-core/src/services/announce.rs index 8d12da713..08ac93f68 100644 --- a/packages/http-tracker-core/src/services/announce.rs +++ b/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); } ( diff --git a/packages/http-tracker-core/src/statistics/event/listener.rs b/packages/http-tracker-core/src/statistics/event/listener.rs index 6730d4c70..ff2937a59 100644 --- a/packages/http-tracker-core/src/statistics/event/listener.rs +++ b/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) -> JoinHandle<()> { +pub fn run_event_listener( + receiver: Receiver, + cancellation_token: CancellationToken, + repository: &Arc, +) -> 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) { - let shutdown_signal = tokio::signal::ctrl_c(); - - tokio::pin!(shutdown_signal); - +async fn dispatch_events(mut receiver: Receiver, cancellation_token: CancellationToken, stats_repository: Arc) { 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 { 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); } } } diff --git a/packages/rest-tracker-api-core/Cargo.toml b/packages/rest-tracker-api-core/Cargo.toml index cc8eda903..be6d493d7 100644 --- a/packages/rest-tracker-api-core/Cargo.toml +++ b/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" } diff --git a/packages/rest-tracker-api-core/src/statistics/services.rs b/packages/rest-tracker-api-core/src/statistics/services.rs index 44c82bfea..a8132d4fd 100644 --- a/packages/rest-tracker-api-core/src/statistics/services.rs +++ b/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 diff --git a/packages/swarm-coordination-registry/Cargo.toml b/packages/swarm-coordination-registry/Cargo.toml index 074562a47..45359ad81 100644 --- a/packages/swarm-coordination-registry/Cargo.toml +++ b/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" } diff --git a/packages/swarm-coordination-registry/src/statistics/event/listener.rs b/packages/swarm-coordination-registry/src/statistics/event/listener.rs index 9ff707818..b578d1284 100644 --- a/packages/swarm-coordination-registry/src/statistics/event/listener.rs +++ b/packages/swarm-coordination-registry/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, SWARM_COORDINATION_REGISTRY_LOG_TARGET}; #[must_use] -pub fn run_event_listener(receiver: Receiver, repository: &Arc) -> JoinHandle<()> { +pub fn run_event_listener( + receiver: Receiver, + cancellation_token: CancellationToken, + repository: &Arc, +) -> JoinHandle<()> { let stats_repository = repository.clone(); - tracing::info!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Starting torrent repository event listener"); + tracing::info!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Starting swarm coordination registry event listener"); tokio::spawn(async move { - dispatch_events(receiver, stats_repository).await; + dispatch_events(receiver, cancellation_token, stats_repository).await; - tracing::info!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Torrent repository listener finished"); + tracing::info!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Swarm coordination registry listener finished"); }) } -async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc) { - let shutdown_signal = tokio::signal::ctrl_c(); - - tokio::pin!(shutdown_signal); - +async fn dispatch_events(mut receiver: Receiver, cancellation_token: CancellationToken, stats_repository: Arc) { loop { tokio::select! { biased; - _ = &mut shutdown_signal => { - tracing::info!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Received Ctrl+C, shutting down torrent repository event listener."); + () = cancellation_token.cancelled() => { + tracing::info!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Received cancellation request, shutting down swarm coordination registry event listener."); break; } @@ -42,11 +43,11 @@ async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc { match e { RecvError::Closed => { - tracing::info!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Torrent repository event receiver closed."); + tracing::info!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Swarm coordination registry event receiver closed."); break; } RecvError::Lagged(n) => { - tracing::warn!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Torrent repository event receiver lagged by {} events.", n); + tracing::warn!(target: SWARM_COORDINATION_REGISTRY_LOG_TARGET, "Swarm coordination registry event receiver lagged by {} events.", n); } } } diff --git a/packages/tracker-core/Cargo.toml b/packages/tracker-core/Cargo.toml index f04a3b89b..dfc83e58e 100644 --- a/packages/tracker-core/Cargo.toml +++ b/packages/tracker-core/Cargo.toml @@ -27,6 +27,7 @@ serde = { version = "1", features = ["derive"] } serde_json = { version = "1", features = ["preserve_order"] } 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" } diff --git a/packages/tracker-core/src/statistics/event/listener.rs b/packages/tracker-core/src/statistics/event/listener.rs index d3beaf41f..8d2d74c71 100644 --- a/packages/tracker-core/src/statistics/event/listener.rs +++ b/packages/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; use torrust_tracker_swarm_coordination_registry::event::receiver::Receiver; @@ -13,6 +14,7 @@ use crate::{CurrentClock, TRACKER_CORE_LOG_TARGET}; #[must_use] pub fn run_event_listener( receiver: Receiver, + cancellation_token: CancellationToken, repository: &Arc, db_downloads_metric_repository: &Arc, persistent_torrent_completed_stat: bool, @@ -20,37 +22,35 @@ pub fn run_event_listener( let stats_repository = repository.clone(); let db_downloads_metric_repository: Arc = db_downloads_metric_repository.clone(); - tracing::info!(target: TRACKER_CORE_LOG_TARGET, "Starting torrent repository event listener"); + tracing::info!(target: TRACKER_CORE_LOG_TARGET, "Starting tracker core event listener"); tokio::spawn(async move { dispatch_events( receiver, + cancellation_token, stats_repository, db_downloads_metric_repository, persistent_torrent_completed_stat, ) .await; - tracing::info!(target: TRACKER_CORE_LOG_TARGET, "Torrent repository listener finished"); + tracing::info!(target: TRACKER_CORE_LOG_TARGET, "Tracker core listener finished"); }) } async fn dispatch_events( mut receiver: Receiver, + cancellation_token: CancellationToken, stats_repository: Arc, db_downloads_metric_repository: Arc, persistent_torrent_completed_stat: bool, ) { - let shutdown_signal = tokio::signal::ctrl_c(); - - tokio::pin!(shutdown_signal); - loop { tokio::select! { biased; - _ = &mut shutdown_signal => { - tracing::info!(target: TRACKER_CORE_LOG_TARGET, "Received Ctrl+C, shutting down torrent repository event listener"); + () = cancellation_token.cancelled() => { + tracing::info!(target: TRACKER_CORE_LOG_TARGET, "Received cancellation request, shutting down tracker core event listener."); break; } @@ -65,11 +65,11 @@ async fn dispatch_events( Err(e) => { match e { RecvError::Closed => { - tracing::info!(target: TRACKER_CORE_LOG_TARGET, "Torrent repository event receiver closed"); + tracing::info!(target: TRACKER_CORE_LOG_TARGET, "Tracker core event receiver closed"); break; } RecvError::Lagged(n) => { - tracing::warn!(target: TRACKER_CORE_LOG_TARGET, "Torrent repository event receiver lagged by {} events", n); + tracing::warn!(target: TRACKER_CORE_LOG_TARGET, "Tracker core event receiver lagged by {} events", n); } } } diff --git a/packages/tracker-core/tests/common/test_env.rs b/packages/tracker-core/tests/common/test_env.rs index d3bc9652a..3fe0464fe 100644 --- a/packages/tracker-core/tests/common/test_env.rs +++ b/packages/tracker-core/tests/common/test_env.rs @@ -7,6 +7,7 @@ use bittorrent_tracker_core::announce_handler::PeersWanted; use bittorrent_tracker_core::container::TrackerCoreContainer; use bittorrent_tracker_core::statistics::persisted::load_persisted_metrics; use tokio::task::yield_now; +use tokio_util::sync::CancellationToken; use torrust_tracker_configuration::Core; use torrust_tracker_metrics::label::LabelSet; use torrust_tracker_metrics::metric::MetricName; @@ -66,15 +67,19 @@ impl TestEnv { async fn run_jobs(&self) { let mut jobs = vec![]; + let cancellation_token = CancellationToken::new(); let job = torrust_tracker_swarm_coordination_registry::statistics::event::listener::run_event_listener( self.swarm_coordination_registry_container.event_bus.receiver(), + cancellation_token.clone(), &self.swarm_coordination_registry_container.stats_repository, ); + jobs.push(job); let job = bittorrent_tracker_core::statistics::event::listener::run_event_listener( self.swarm_coordination_registry_container.event_bus.receiver(), + cancellation_token.clone(), &self.tracker_core_container.stats_repository, &self.tracker_core_container.db_downloads_metric_repository, self.tracker_core_container diff --git a/packages/udp-tracker-core/Cargo.toml b/packages/udp-tracker-core/Cargo.toml index 290c5fbfd..b3007eb80 100644 --- a/packages/udp-tracker-core/Cargo.toml +++ b/packages/udp-tracker-core/Cargo.toml @@ -28,6 +28,7 @@ rand = "0" serde = "1.0.219" thiserror = "2" tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"] } +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" } diff --git a/packages/udp-tracker-core/src/statistics/event/listener.rs b/packages/udp-tracker-core/src/statistics/event/listener.rs index 9b6f2e574..b11bcce85 100644 --- a/packages/udp-tracker-core/src/statistics/event/listener.rs +++ b/packages/udp-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,28 +11,29 @@ use crate::statistics::repository::Repository; use crate::{CurrentClock, UDP_TRACKER_LOG_TARGET}; #[must_use] -pub fn run_event_listener(receiver: Receiver, repository: &Arc) -> JoinHandle<()> { +pub fn run_event_listener( + receiver: Receiver, + cancellation_token: CancellationToken, + repository: &Arc, +) -> JoinHandle<()> { let stats_repository = repository.clone(); tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Starting UDP 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: UDP_TRACKER_LOG_TARGET, "UDP tracker core event listener finished"); }) } -async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc) { - let shutdown_signal = tokio::signal::ctrl_c(); - tokio::pin!(shutdown_signal); - +async fn dispatch_events(mut receiver: Receiver, cancellation_token: CancellationToken, stats_repository: Arc) { loop { tokio::select! { biased; - _ = &mut shutdown_signal => { - tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Received Ctrl+C, shutting down UDP tracker core event listener."); + () = cancellation_token.cancelled() => { + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Received cancellation request, shutting down UDP tracker core event listener."); break; } @@ -41,11 +43,11 @@ async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc { match e { RecvError::Closed => { - tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp core statistics receiver closed."); + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp tracker core statistics receiver closed."); break; } RecvError::Lagged(n) => { - tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp core statistics receiver lagged by {} events.", n); + tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp tracker core statistics receiver lagged by {} events.", n); } } } diff --git a/packages/udp-tracker-server/Cargo.toml b/packages/udp-tracker-server/Cargo.toml index c0bc94ce3..160fe58f9 100644 --- a/packages/udp-tracker-server/Cargo.toml +++ b/packages/udp-tracker-server/Cargo.toml @@ -26,6 +26,7 @@ ringbuf = "0" serde = "1.0.219" thiserror = "2" tokio = { version = "1", features = ["macros", "net", "rt-multi-thread", "signal", "sync"] } +tokio-util = "0.7.15" torrust-server-lib = { version = "3.0.0-develop", path = "../server-lib" } torrust-tracker-clock = { version = "3.0.0-develop", path = "../clock" } torrust-tracker-configuration = { version = "3.0.0-develop", path = "../configuration" } diff --git a/packages/udp-tracker-server/src/banning/event/listener.rs b/packages/udp-tracker-server/src/banning/event/listener.rs index fee3395fa..0d579f912 100644 --- a/packages/udp-tracker-server/src/banning/event/listener.rs +++ b/packages/udp-tracker-server/src/banning/event/listener.rs @@ -4,6 +4,7 @@ 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 tokio_util::sync::CancellationToken; use torrust_tracker_clock::clock::Time; use torrust_tracker_events::receiver::RecvError; @@ -15,6 +16,7 @@ use crate::CurrentClock; #[must_use] pub fn run_event_listener( receiver: Receiver, + cancellation_token: CancellationToken, ban_service: &Arc>, repository: &Arc, ) -> JoinHandle<()> { @@ -24,22 +26,24 @@ pub fn run_event_listener( tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Starting UDP tracker server event listener (banning)"); tokio::spawn(async move { - dispatch_events(receiver, ban_service_clone, repository_clone).await; + dispatch_events(receiver, cancellation_token, ban_service_clone, repository_clone).await; tracing::info!(target: UDP_TRACKER_LOG_TARGET, "UDP tracker server event listener (banning) finished"); }) } -async fn dispatch_events(mut receiver: Receiver, ban_service: Arc>, repository: Arc) { - let shutdown_signal = tokio::signal::ctrl_c(); - tokio::pin!(shutdown_signal); - +async fn dispatch_events( + mut receiver: Receiver, + cancellation_token: CancellationToken, + ban_service: Arc>, + repository: Arc, +) { loop { tokio::select! { biased; - _ = &mut shutdown_signal => { - tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Received Ctrl+C, shutting down UDP tracker server event listener (banning)"); + () = cancellation_token.cancelled() => { + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Received cancellation request, shutting down UDP tracker server event listener."); break; } @@ -49,11 +53,11 @@ async fn dispatch_events(mut receiver: Receiver, ban_service: Arc { match e { RecvError::Closed => { - tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp server receiver (banning) closed."); + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp tracker server receiver (banning) closed."); break; } RecvError::Lagged(n) => { - tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp server receiver (banning) lagged by {} events.", n); + tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp tracker server receiver (banning) lagged by {} events.", n); } } } diff --git a/packages/udp-tracker-server/src/environment.rs b/packages/udp-tracker-server/src/environment.rs index 61b1cba63..13e18ba9b 100644 --- a/packages/udp-tracker-server/src/environment.rs +++ b/packages/udp-tracker-server/src/environment.rs @@ -4,6 +4,7 @@ use std::sync::Arc; use bittorrent_tracker_core::container::TrackerCoreContainer; use bittorrent_udp_tracker_core::container::UdpTrackerCoreContainer; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use torrust_server_lib::registar::Registar; use torrust_tracker_configuration::{logging, Configuration, DEFAULT_TIMEOUT}; use torrust_tracker_swarm_coordination_registry::container::SwarmCoordinationRegistryContainer; @@ -25,6 +26,7 @@ where pub udp_core_event_listener_job: Option>, pub udp_server_stats_event_listener_job: Option>, pub udp_server_banning_event_listener_job: Option>, + pub cancellation_token: CancellationToken, } impl Environment { @@ -46,6 +48,7 @@ impl Environment { udp_core_event_listener_job: None, udp_server_stats_event_listener_job: None, udp_server_banning_event_listener_job: None, + cancellation_token: CancellationToken::new(), } } @@ -57,21 +60,25 @@ impl Environment { #[allow(dead_code)] pub async fn start(self) -> Environment { let cookie_lifetime = self.container.udp_tracker_core_container.udp_tracker_config.cookie_lifetime; + // Start the UDP tracker core event listener let udp_core_event_listener_job = Some(bittorrent_udp_tracker_core::statistics::event::listener::run_event_listener( self.container.udp_tracker_core_container.event_bus.receiver(), + self.cancellation_token.clone(), &self.container.udp_tracker_core_container.stats_repository, )); // Start the UDP tracker server event listener (statistics) let udp_server_stats_event_listener_job = Some(crate::statistics::event::listener::run_event_listener( self.container.udp_tracker_server_container.event_bus.receiver(), + self.cancellation_token.clone(), &self.container.udp_tracker_server_container.stats_repository, )); // Start the UDP tracker server event listener (banning) let udp_server_banning_event_listener_job = Some(crate::banning::event::listener::run_event_listener( self.container.udp_tracker_server_container.event_bus.receiver(), + self.cancellation_token.clone(), &self.container.udp_tracker_core_container.ban_service, &self.container.udp_tracker_server_container.stats_repository, )); @@ -95,6 +102,7 @@ impl Environment { udp_core_event_listener_job, udp_server_stats_event_listener_job, udp_server_banning_event_listener_job, + cancellation_token: self.cancellation_token, } } } @@ -150,6 +158,7 @@ impl Environment { udp_core_event_listener_job: None, udp_server_stats_event_listener_job: None, udp_server_banning_event_listener_job: None, + cancellation_token: self.cancellation_token, } } diff --git a/packages/udp-tracker-server/src/statistics/event/listener.rs b/packages/udp-tracker-server/src/statistics/event/listener.rs index ae659c15e..caaf5a2bc 100644 --- a/packages/udp-tracker-server/src/statistics/event/listener.rs +++ b/packages/udp-tracker-server/src/statistics/event/listener.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use bittorrent_udp_tracker_core::UDP_TRACKER_LOG_TARGET; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use torrust_tracker_clock::clock::Time; use torrust_tracker_events::receiver::RecvError; @@ -11,28 +12,29 @@ use crate::statistics::repository::Repository; use crate::CurrentClock; #[must_use] -pub fn run_event_listener(receiver: Receiver, repository: &Arc) -> JoinHandle<()> { +pub fn run_event_listener( + receiver: Receiver, + cancellation_token: CancellationToken, + repository: &Arc, +) -> JoinHandle<()> { let repository_clone = repository.clone(); tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Starting UDP tracker server event listener"); tokio::spawn(async move { - dispatch_events(receiver, repository_clone).await; + dispatch_events(receiver, cancellation_token, repository_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) { - let shutdown_signal = tokio::signal::ctrl_c(); - tokio::pin!(shutdown_signal); - +async fn dispatch_events(mut receiver: Receiver, cancellation_token: CancellationToken, stats_repository: Arc) { loop { tokio::select! { biased; - _ = &mut shutdown_signal => { - tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Received Ctrl+C, shutting down UDP tracker server event listener."); + () = cancellation_token.cancelled() => { + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Received cancellation request, shutting down UDP tracker server event listener."); break; } @@ -42,11 +44,11 @@ async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc { match e { RecvError::Closed => { - tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp server statistics receiver closed."); + tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp tracker server statistics receiver closed."); break; } RecvError::Lagged(n) => { - tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp server statistics receiver lagged by {} events.", n); + tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp tracker server statistics receiver lagged by {} events.", n); } } } diff --git a/src/app.rs b/src/app.rs index 58d758d7f..2149a6d4c 100644 --- a/src/app.rs +++ b/src/app.rs @@ -140,28 +140,28 @@ fn start_swarm_coordination_registry_event_listener( ) { job_manager.push_opt( "swarm_coordination_registry_event_listener", - jobs::torrent_repository::start_event_listener(config, app_container), + jobs::torrent_repository::start_event_listener(config, app_container, job_manager.new_cancellation_token()), ); } fn start_tracker_core_event_listener(config: &Configuration, app_container: &Arc, job_manager: &mut JobManager) { job_manager.push_opt( "tracker_core_event_listener", - jobs::tracker_core::start_event_listener(config, app_container), + jobs::tracker_core::start_event_listener(config, app_container, job_manager.new_cancellation_token()), ); } fn start_http_core_event_listener(config: &Configuration, app_container: &Arc, job_manager: &mut JobManager) { job_manager.push_opt( "http_core_event_listener", - jobs::http_tracker_core::start_event_listener(config, app_container), + jobs::http_tracker_core::start_event_listener(config, app_container, job_manager.new_cancellation_token()), ); } fn start_udp_core_event_listener(config: &Configuration, app_container: &Arc, job_manager: &mut JobManager) { job_manager.push_opt( "udp_core_event_listener", - jobs::udp_tracker_core::start_event_listener(config, app_container), + jobs::udp_tracker_core::start_event_listener(config, app_container, job_manager.new_cancellation_token()), ); } @@ -172,14 +172,14 @@ fn start_udp_server_stats_event_listener( ) { job_manager.push_opt( "udp_server_stats_event_listener", - jobs::udp_tracker_server::start_stats_event_listener(config, app_container), + jobs::udp_tracker_server::start_stats_event_listener(config, app_container, job_manager.new_cancellation_token()), ); } fn start_udp_server_banning_event_listener(app_container: &Arc, job_manager: &mut JobManager) { job_manager.push( "udp_server_banning_event_listener", - jobs::udp_tracker_server::start_banning_event_listener(app_container), + jobs::udp_tracker_server::start_banning_event_listener(app_container, job_manager.new_cancellation_token()), ); } diff --git a/src/bootstrap/jobs/http_tracker_core.rs b/src/bootstrap/jobs/http_tracker_core.rs index 952c80b40..ab71b9a0f 100644 --- a/src/bootstrap/jobs/http_tracker_core.rs +++ b/src/bootstrap/jobs/http_tracker_core.rs @@ -1,14 +1,20 @@ use std::sync::Arc; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use torrust_tracker_configuration::Configuration; use crate::container::AppContainer; -pub fn start_event_listener(config: &Configuration, app_container: &Arc) -> Option> { +pub fn start_event_listener( + config: &Configuration, + app_container: &Arc, + cancellation_token: CancellationToken, +) -> Option> { if config.core.tracker_usage_statistics { let job = bittorrent_http_tracker_core::statistics::event::listener::run_event_listener( app_container.http_tracker_core_services.event_bus.receiver(), + cancellation_token, &app_container.http_tracker_core_services.stats_repository, ); diff --git a/src/bootstrap/jobs/manager.rs b/src/bootstrap/jobs/manager.rs index 53733844b..565cd7b73 100644 --- a/src/bootstrap/jobs/manager.rs +++ b/src/bootstrap/jobs/manager.rs @@ -2,13 +2,14 @@ use std::time::Duration; use tokio::task::JoinHandle; use tokio::time::timeout; +use tokio_util::sync::CancellationToken; use tracing::{info, warn}; /// Represents a named background job. #[derive(Debug)] pub struct Job { - pub name: String, - pub handle: JoinHandle<()>, + name: String, + handle: JoinHandle<()>, } impl Job { @@ -24,12 +25,16 @@ impl Job { #[derive(Debug, Default)] pub struct JobManager { jobs: Vec, + cancellation_token: CancellationToken, } impl JobManager { #[must_use] pub fn new() -> Self { - Self { jobs: Vec::new() } + Self { + jobs: Vec::new(), + cancellation_token: CancellationToken::new(), + } } pub fn push>(&mut self, name: N, handle: JoinHandle<()>) { @@ -42,6 +47,25 @@ impl JobManager { } } + #[must_use] + pub fn new_cancellation_token(&self) -> CancellationToken { + self.cancellation_token.clone() + } + + /// Cancels all jobs using the shared cancellation token. + /// + /// Notice that this does not cancel the jobs immediately, but rather + /// signals them to stop. The jobs themselves must handle the cancellation + /// token appropriately. + /// + /// Notice jobs might be pushed into the manager without a cancellation + /// token, so this method will not cancel those jobs. Some tasks might + /// decide to listen for CTRL+c signal directly, or implement their own + /// cancellation logic. + pub fn cancel(&self) { + self.cancellation_token.cancel(); + } + /// Waits sequentially for all jobs to complete, with a graceful timeout per /// job. pub async fn wait_for_all(mut self, grace_period: Duration) { diff --git a/src/bootstrap/jobs/torrent_repository.rs b/src/bootstrap/jobs/torrent_repository.rs index 44ffdf53b..e49323735 100644 --- a/src/bootstrap/jobs/torrent_repository.rs +++ b/src/bootstrap/jobs/torrent_repository.rs @@ -1,14 +1,20 @@ use std::sync::Arc; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use torrust_tracker_configuration::Configuration; use crate::container::AppContainer; -pub fn start_event_listener(config: &Configuration, app_container: &Arc) -> Option> { +pub fn start_event_listener( + config: &Configuration, + app_container: &Arc, + cancellation_token: CancellationToken, +) -> Option> { if config.core.tracker_usage_statistics { let job = torrust_tracker_swarm_coordination_registry::statistics::event::listener::run_event_listener( app_container.swarm_coordination_registry_container.event_bus.receiver(), + cancellation_token, &app_container.swarm_coordination_registry_container.stats_repository, ); diff --git a/src/bootstrap/jobs/tracker_core.rs b/src/bootstrap/jobs/tracker_core.rs index f2fc25ef3..d881f4cd2 100644 --- a/src/bootstrap/jobs/tracker_core.rs +++ b/src/bootstrap/jobs/tracker_core.rs @@ -1,14 +1,20 @@ use std::sync::Arc; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use torrust_tracker_configuration::Configuration; use crate::container::AppContainer; -pub fn start_event_listener(config: &Configuration, app_container: &Arc) -> Option> { +pub fn start_event_listener( + config: &Configuration, + app_container: &Arc, + cancellation_token: CancellationToken, +) -> Option> { if config.core.tracker_usage_statistics || config.core.tracker_policy.persistent_torrent_completed_stat { let job = bittorrent_tracker_core::statistics::event::listener::run_event_listener( app_container.swarm_coordination_registry_container.event_bus.receiver(), + cancellation_token, &app_container.tracker_core_container.stats_repository, &app_container.tracker_core_container.db_downloads_metric_repository, app_container diff --git a/src/bootstrap/jobs/udp_tracker_core.rs b/src/bootstrap/jobs/udp_tracker_core.rs index 689fa8301..dd7e8c165 100644 --- a/src/bootstrap/jobs/udp_tracker_core.rs +++ b/src/bootstrap/jobs/udp_tracker_core.rs @@ -1,14 +1,20 @@ use std::sync::Arc; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use torrust_tracker_configuration::Configuration; use crate::container::AppContainer; -pub fn start_event_listener(config: &Configuration, app_container: &Arc) -> Option> { +pub fn start_event_listener( + config: &Configuration, + app_container: &Arc, + cancellation_token: CancellationToken, +) -> Option> { if config.core.tracker_usage_statistics { let job = bittorrent_udp_tracker_core::statistics::event::listener::run_event_listener( app_container.udp_tracker_core_services.event_bus.receiver(), + cancellation_token, &app_container.udp_tracker_core_services.stats_repository, ); Some(job) diff --git a/src/bootstrap/jobs/udp_tracker_server.rs b/src/bootstrap/jobs/udp_tracker_server.rs index 3e8a7aaa8..fc6df9c16 100644 --- a/src/bootstrap/jobs/udp_tracker_server.rs +++ b/src/bootstrap/jobs/udp_tracker_server.rs @@ -1,14 +1,20 @@ use std::sync::Arc; use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; use torrust_tracker_configuration::Configuration; use crate::container::AppContainer; -pub fn start_stats_event_listener(config: &Configuration, app_container: &Arc) -> Option> { +pub fn start_stats_event_listener( + config: &Configuration, + app_container: &Arc, + cancellation_token: CancellationToken, +) -> Option> { if config.core.tracker_usage_statistics { let job = torrust_udp_tracker_server::statistics::event::listener::run_event_listener( app_container.udp_tracker_server_container.event_bus.receiver(), + cancellation_token, &app_container.udp_tracker_server_container.stats_repository, ); Some(job) @@ -19,9 +25,10 @@ pub fn start_stats_event_listener(config: &Configuration, app_container: &Arc) -> JoinHandle<()> { +pub fn start_banning_event_listener(app_container: &Arc, cancellation_token: CancellationToken) -> JoinHandle<()> { torrust_udp_tracker_server::banning::event::listener::run_event_listener( app_container.udp_tracker_server_container.event_bus.receiver(), + cancellation_token, &app_container.udp_tracker_core_services.ban_service, &app_container.udp_tracker_server_container.stats_repository, ) diff --git a/src/main.rs b/src/main.rs index a49c3aeba..7012ecaa7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,6 +10,8 @@ async fn main() { _ = tokio::signal::ctrl_c() => { tracing::info!("Torrust tracker shutting down ..."); + jobs.cancel(); + jobs.wait_for_all(Duration::from_secs(10)).await; tracing::info!("Torrust tracker successfully shutdown.");