feat: [#1477] extract JobManager to handle jobs

It:

- Give a name to all jobs so they can be identify later in logs.
- Wait for all jobs to finish when the app receives teh halt signal
  (CRTL+c)
- Only waits for a grace period per job.
- Shows a message when a job don't complete in time.

This could be improved in the future:

- By showing a message every second while we are waiting for a job to finish.
- Waiting for all of them in paralell.

Author: josecelano

src/app.rs

@@ -23,15 +23,15 @@
 //! - Tracker REST API: the tracker API can be enabled/disabled.
 use std::sync::Arc;
 
-use tokio::task::JoinHandle;
 use torrust_tracker_configuration::{Configuration, HttpTracker, UdpTracker};
 use tracing::instrument;
 
+use crate::bootstrap::jobs::manager::JobManager;
 use crate::bootstrap::jobs::{self, health_check_api, http_tracker, torrent_cleanup, tracker_apis, udp_tracker};
 use crate::bootstrap::{self};
 use crate::container::AppContainer;
 
-pub async fn run() -> (Arc<AppContainer>, Vec<JoinHandle<()>>) {
+pub async fn run() -> (Arc<AppContainer>, JobManager) {
     let (config, app_container) = bootstrap::app::setup();
 
     let app_container = Arc::new(app_container);
@@ -50,7 +50,7 @@ pub async fn run() -> (Arc<AppContainer>, Vec<JoinHandle<()>>) {
 /// - Can't retrieve tracker keys from database.
 /// - Can't load whitelist from database.
 #[instrument(skip(config, app_container))]
-pub async fn start(config: &Configuration, app_container: &Arc<AppContainer>) -> Vec<JoinHandle<()>> {
+pub async fn start(config: &Configuration, app_container: &Arc<AppContainer>) -> JobManager {
     warn_if_no_services_enabled(config);
 
     load_data_from_database(config, app_container).await;
@@ -63,19 +63,19 @@ async fn load_data_from_database(config: &Configuration, app_container: &Arc<App
     load_whitelisted_torrents(config, app_container).await;
 }
 
-async fn start_jobs(config: &Configuration, app_container: &Arc<AppContainer>) -> Vec<JoinHandle<()>> {
-    let mut jobs: Vec<JoinHandle<()>> = Vec::new();
+async fn start_jobs(config: &Configuration, app_container: &Arc<AppContainer>) -> JobManager {
+    let mut job_manager = JobManager::new();
 
-    start_http_core_event_listener(config, app_container, &mut jobs);
-    start_udp_core_event_listener(config, app_container, &mut jobs);
-    start_udp_server_event_listener(config, app_container, &mut jobs);
-    start_the_udp_instances(config, app_container, &mut jobs).await;
-    start_the_http_instances(config, app_container, &mut jobs).await;
-    start_the_http_api(config, app_container, &mut jobs).await;
-    start_torrent_cleanup(config, app_container, &mut jobs);
-    start_health_check_api(config, app_container, &mut jobs).await;
+    start_http_core_event_listener(config, app_container, &mut job_manager);
+    start_udp_core_event_listener(config, app_container, &mut job_manager);
+    start_udp_server_event_listener(config, app_container, &mut job_manager);
+    start_the_udp_instances(config, app_container, &mut job_manager).await;
+    start_the_http_instances(config, app_container, &mut job_manager).await;
+    start_the_http_api(config, app_container, &mut job_manager).await;
+    start_torrent_cleanup(config, app_container, &mut job_manager);
+    start_health_check_api(config, app_container, &mut job_manager).await;
 
-    jobs
+    job_manager
 }
 
 fn warn_if_no_services_enabled(config: &Configuration) {
@@ -109,94 +109,100 @@ async fn load_whitelisted_torrents(config: &Configuration, app_container: &Arc<A
     }
 }
 
-fn start_http_core_event_listener(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
-    let opt_job = jobs::http_tracker_core::start_event_listener(config, app_container);
+fn start_http_core_event_listener(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
+    let opt_handle = jobs::http_tracker_core::start_event_listener(config, app_container);
 
-    if let Some(job) = opt_job {
-        jobs.push(job);
+    if let Some(handle) = opt_handle {
+        job_manager.push("http_core_event_listener", handle);
     }
 }
 
-fn start_udp_core_event_listener(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
-    let opt_job = jobs::udp_tracker_core::start_event_listener(config, app_container);
+fn start_udp_core_event_listener(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
+    let opt_handle = jobs::udp_tracker_core::start_event_listener(config, app_container);
 
-    if let Some(job) = opt_job {
-        jobs.push(job);
+    if let Some(handle) = opt_handle {
+        job_manager.push("udp_core_event_listener", handle);
     }
 }
 
-fn start_udp_server_event_listener(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
-    let opt_job = jobs::udp_tracker_server::start_event_listener(config, app_container);
+fn start_udp_server_event_listener(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
+    let opt_handle = jobs::udp_tracker_server::start_event_listener(config, app_container);
 
-    if let Some(job) = opt_job {
-        jobs.push(job);
+    if let Some(handle) = opt_handle {
+        job_manager.push("udp_server_event_listener", handle);
     }
 }
 
-async fn start_the_udp_instances(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
+async fn start_the_udp_instances(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
     if let Some(udp_trackers) = &config.udp_trackers {
-        for udp_tracker_config in udp_trackers {
+        for (idx, udp_tracker_config) in udp_trackers.iter().enumerate() {
             if config.core.private {
                 tracing::warn!(
                     "Could not start UDP tracker on: {} while in private mode. UDP is not safe for private trackers!",
                     udp_tracker_config.bind_address
                 );
             } else {
-                start_udp_instance(udp_tracker_config, app_container, jobs).await;
+                start_udp_instance(idx, udp_tracker_config, app_container, job_manager).await;
             }
         }
     } else {
         tracing::info!("No UDP blocks in configuration");
     }
 }
 
-async fn start_udp_instance(udp_tracker_config: &UdpTracker, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
+async fn start_udp_instance(
+    idx: usize,
+    udp_tracker_config: &UdpTracker,
+    app_container: &Arc<AppContainer>,
+    job_manager: &mut JobManager,
+) {
     let udp_tracker_container = app_container
         .udp_tracker_container(udp_tracker_config.bind_address)
         .expect("Could not create UDP tracker container");
     let udp_tracker_server_container = app_container.udp_tracker_server_container();
 
-    jobs.push(
-        udp_tracker::start_job(
-            udp_tracker_container,
-            udp_tracker_server_container,
-            app_container.registar.give_form(),
-        )
-        .await,
-    );
+    let handle = udp_tracker::start_job(
+        udp_tracker_container,
+        udp_tracker_server_container,
+        app_container.registar.give_form(),
+    )
+    .await;
+
+    job_manager.push(format!("udp_instance_{}_{}", idx, udp_tracker_config.bind_address), handle);
 }
 
-async fn start_the_http_instances(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
+async fn start_the_http_instances(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
     if let Some(http_trackers) = &config.http_trackers {
-        for http_tracker_config in http_trackers {
-            start_http_instance(http_tracker_config, app_container, jobs).await;
+        for (idx, http_tracker_config) in http_trackers.iter().enumerate() {
+            start_http_instance(idx, http_tracker_config, app_container, job_manager).await;
         }
     } else {
         tracing::info!("No HTTP blocks in configuration");
     }
 }
 
 async fn start_http_instance(
+    idx: usize,
     http_tracker_config: &HttpTracker,
     app_container: &Arc<AppContainer>,
-    jobs: &mut Vec<JoinHandle<()>>,
+    job_manager: &mut JobManager,
 ) {
     let http_tracker_container = app_container
         .http_tracker_container(http_tracker_config.bind_address)
         .expect("Could not create HTTP tracker container");
 
-    if let Some(job) = http_tracker::start_job(
+    if let Some(handle) = http_tracker::start_job(
         http_tracker_container,
         app_container.registar.give_form(),
         torrust_axum_http_tracker_server::Version::V1,
     )
     .await
     {
-        jobs.push(job);
+        job_manager.push(format!("http_instance_{}_{}", idx, http_tracker_config.bind_address), handle);
     }
 }
 
-async fn start_the_http_api(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
+async fn start_the_http_api(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
     if let Some(http_api_config) = &config.http_api {
         let http_api_config = Arc::new(http_api_config.clone());
         let http_api_container = app_container.tracker_http_api_container(&http_api_config);
@@ -208,22 +214,23 @@ async fn start_the_http_api(config: &Configuration, app_container: &Arc<AppConta
         )
         .await
         {
-            jobs.push(job);
+            job_manager.push("http_api", job);
         }
     } else {
         tracing::info!("No API block in configuration");
     }
 }
 
-fn start_torrent_cleanup(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
+fn start_torrent_cleanup(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
     if config.core.inactive_peer_cleanup_interval > 0 {
-        jobs.push(torrent_cleanup::start_job(
-            &config.core,
-            &app_container.tracker_core_container.torrents_manager,
-        ));
+        let handle = torrent_cleanup::start_job(&config.core, &app_container.tracker_core_container.torrents_manager);
+
+        job_manager.push("torrent_cleanup", handle);
     }
 }
 
-async fn start_health_check_api(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {
-    jobs.push(health_check_api::start_job(&config.health_check_api, app_container.registar.entries()).await);
+async fn start_health_check_api(config: &Configuration, app_container: &Arc<AppContainer>, job_manager: &mut JobManager) {
+    let handle = health_check_api::start_job(&config.health_check_api, app_container.registar.entries()).await;
+
+    job_manager.push("health_check_api", handle);
 }

src/bootstrap/jobs/manager.rs

@@ -0,0 +1,89 @@
+use std::time::Duration;
+
+use tokio::task::JoinHandle;
+use tokio::time::timeout;
+use tracing::{info, warn};
+
+/// Represents a named background job.
+#[derive(Debug)]
+pub struct Job {
+    pub name: String,
+    pub handle: JoinHandle<()>,
+}
+
+impl Job {
+    pub fn new<N: Into<String>>(name: N, handle: JoinHandle<()>) -> Self {
+        Self {
+            name: name.into(),
+            handle,
+        }
+    }
+}
+
+/// Manages multiple background jobs.
+#[derive(Debug, Default)]
+pub struct JobManager {
+    jobs: Vec<Job>,
+}
+
+impl JobManager {
+    #[must_use]
+    pub fn new() -> Self {
+        Self { jobs: Vec::new() }
+    }
+
+    pub fn push<N: Into<String>>(&mut self, name: N, handle: JoinHandle<()>) {
+        self.jobs.push(Job::new(name, handle));
+    }
+
+    /// Waits sequentially for all jobs to complete, with a graceful timeout per
+    /// job.
+    pub async fn wait_for_all(mut self, grace_period: Duration) {
+        for job in self.jobs.drain(..) {
+            let name = job.name.clone();
+
+            info!(job = %name, "Waiting for job to finish (timeout of {} seconds) ...", grace_period.as_secs());
+
+            if let Ok(result) = timeout(grace_period, job.handle).await {
+                if let Err(e) = result {
+                    warn!(job = %name, "Job return an error: {:?}", e);
+                } else {
+                    info!(job = %name, "Job completed gracefully");
+                }
+            } else {
+                warn!(job = %name, "Job did not complete in time");
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use tokio::time::Duration;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn it_should_wait_for_all_jobs_to_finish() {
+        let mut manager = JobManager::new();
+
+        manager.push("job1", tokio::spawn(async {}));
+        manager.push("job2", tokio::spawn(async {}));
+
+        manager.wait_for_all(Duration::from_secs(1)).await;
+    }
+
+    #[tokio::test]
+    async fn it_should_log_when_a_job_panics() {
+        let mut manager = JobManager::new();
+
+        manager.push(
+            "panic_job",
+            tokio::spawn(async {
+                panic!("expected panic");
+            }),
+        );
+
+        manager.wait_for_all(Duration::from_secs(1)).await;
+    }
+}

src/bootstrap/jobs/mod.rs

@@ -9,6 +9,7 @@
 pub mod health_check_api;
 pub mod http_tracker;
 pub mod http_tracker_core;
+pub mod manager;
 pub mod torrent_cleanup;
 pub mod tracker_apis;
 pub mod udp_tracker;

src/console/profiling.rs

@@ -191,8 +191,7 @@ pub async fn run() {
         _ = tokio::signal::ctrl_c() => {
             tracing::info!("Torrust tracker shutting down via Ctrl+C ...");
 
-            // Await for all jobs to shutdown
-            futures::future::join_all(jobs).await;
+            jobs.wait_for_all(Duration::from_secs(10)).await;
         }
     }
 

src/main.rs

@@ -1,16 +1,16 @@
+use std::time::Duration;
+
 use torrust_tracker_lib::app;
 
 #[tokio::main]
 async fn main() {
     let (_app_container, jobs) = app::run().await;
 
-    // handle the signals
     tokio::select! {
         _ = tokio::signal::ctrl_c() => {
             tracing::info!("Torrust tracker shutting down ...");
 
-            // Await for all jobs to shutdown
-            futures::future::join_all(jobs).await;
+            jobs.wait_for_all(Duration::from_secs(10)).await;
 
             tracing::info!("Torrust tracker successfully shutdown.");
         }