fix: [torrust#1477] shutdown event listeners on CRTL+c signal

This fixes the problem of adding the jobs for event listeners in the
main app JoinHandle vector. It was not possible to add the handles for
those tokio tasks becuase in the main app we wait for all jobs and those
jobs never end.

```rust
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;

            tracing::info!("Torrust tracker successfully shutdown.");
        }
    }
}
```

Now, we can wait for them becuase they listen for the halt signal.

We will implement the shutdown in a different way in a new PR. See

torrust#1405

Instead of listen to the CRTL+c signal the main app will send a "stop"
event to the listeners. The final goal it only the main app listen for
this external signal and it propagates the shutdown in cascade via
normal internal messages or channels.

Author: josecelano

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

@@ -23,18 +23,32 @@ pub fn run_event_listener(receiver: Receiver, repository: &Arc<Repository>) -> J
 }
 
 async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repository>) {
+    let shutdown_signal = tokio::signal::ctrl_c();
+
+    tokio::pin!(shutdown_signal);
+
     loop {
-        match receiver.recv().await {
-            Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
-            Err(e) => {
-                match e {
-                    RecvError::Closed => {
-                        tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Http core statistics receiver closed.");
-                        break;
-                    }
-                    RecvError::Lagged(n) => {
-                        // From now on, metrics will be imprecise
-                        tracing::warn!(target: HTTP_TRACKER_LOG_TARGET, "Http core statistics receiver lagged by {} events.", n);
+        tokio::select! {
+            biased;
+
+            _ = &mut shutdown_signal => {
+                tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Received Ctrl+C, shutting down HTTP tracker core event listener.");
+                break;
+            }
+
+            result = receiver.recv() => {
+                match result {
+                    Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
+                    Err(e) => {
+                        match e {
+                            RecvError::Closed => {
+                                tracing::info!(target: HTTP_TRACKER_LOG_TARGET, "Http core statistics receiver closed.");
+                                break;
+                            }
+                            RecvError::Lagged(n) => {
+                                tracing::warn!(target: HTTP_TRACKER_LOG_TARGET, "Http core statistics receiver lagged by {} events.", n);
+                            }
+                        }
                     }
                 }
             }

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

@@ -23,18 +23,31 @@ pub fn run_event_listener(receiver: Receiver, repository: &Arc<Repository>) -> J
 }
 
 async fn dispatch_events(mut receiver: Receiver, stats_repository: Arc<Repository>) {
+    let shutdown_signal = tokio::signal::ctrl_c();
+    tokio::pin!(shutdown_signal);
+
     loop {
-        match receiver.recv().await {
-            Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
-            Err(e) => {
-                match e {
-                    RecvError::Closed => {
-                        tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp core statistics receiver closed.");
-                        break;
-                    }
-                    RecvError::Lagged(n) => {
-                        // From now on, metrics will be imprecise
-                        tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp core statistics receiver lagged by {} events.", n);
+        tokio::select! {
+            biased;
+
+            _ = &mut shutdown_signal => {
+                tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Received Ctrl+C, shutting down UDP tracker core event listener.");
+                break;
+            }
+
+            result = receiver.recv() => {
+                match result {
+                    Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
+                    Err(e) => {
+                        match e {
+                            RecvError::Closed => {
+                                tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp core statistics receiver closed.");
+                                break;
+                            }
+                            RecvError::Lagged(n) => {
+                                tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp core statistics receiver lagged by {} events.", n);
+                            }
+                        }
                     }
                 }
             }

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

@@ -19,23 +19,36 @@ pub fn run_event_listener(receiver: Receiver, repository: &Arc<Repository>) -> J
     tokio::spawn(async move {
         dispatch_events(receiver, stats_repository).await;
 
-        tracing::info!(target: UDP_TRACKER_LOG_TARGET, "DP tracker server event listener finished");
+        tracing::info!(target: UDP_TRACKER_LOG_TARGET, "UDP tracker server 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);
+
     loop {
-        match receiver.recv().await {
-            Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
-            Err(e) => {
-                match e {
-                    RecvError::Closed => {
-                        tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp server statistics receiver closed.");
-                        break;
-                    }
-                    RecvError::Lagged(n) => {
-                        // From now on, metrics will be imprecise
-                        tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp server statistics receiver lagged by {} events.", n);
+        tokio::select! {
+            biased;
+
+            _ = &mut shutdown_signal => {
+                tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Received Ctrl+C, shutting down UDP tracker server event listener.");
+                break;
+            }
+
+            result = receiver.recv() => {
+                match result {
+                    Ok(event) => handle_event(event, &stats_repository, CurrentClock::now()).await,
+                    Err(e) => {
+                        match e {
+                            RecvError::Closed => {
+                                tracing::info!(target: UDP_TRACKER_LOG_TARGET, "Udp server statistics receiver closed.");
+                                break;
+                            }
+                            RecvError::Lagged(n) => {
+                                tracing::warn!(target: UDP_TRACKER_LOG_TARGET, "Udp server statistics receiver lagged by {} events.", n);
+                            }
+                        }
                     }
                 }
             }

src/app.rs

@@ -66,9 +66,9 @@ async fn load_data_from_database(config: &Configuration, app_container: &Arc<App
 async fn start_jobs(config: &Configuration, app_container: &Arc<AppContainer>) -> Vec<JoinHandle<()>> {
     let mut jobs: Vec<JoinHandle<()>> = Vec::new();
 
-    start_http_core_event_listener(config, app_container);
-    start_udp_core_event_listener(config, app_container);
-    start_udp_server_event_listener(config, app_container);
+    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;
@@ -109,43 +109,28 @@ async fn load_whitelisted_torrents(config: &Configuration, app_container: &Arc<A
     }
 }
 
-fn start_http_core_event_listener(config: &Configuration, app_container: &Arc<AppContainer>) {
-    let _job = jobs::http_tracker_core::start_event_listener(config, app_container);
+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);
 
-    // todo: this cannot be enabled otherwise the application never ends
-    // because the event listener never stops. You see this console message
-    // forever:
-    //
-    // !! shuting down in 90 seconds !!
-    // 2025-04-24T15:27:45.454101Z  INFO graceful_shutdown: torrust_axum_server::signals: remaining alive connections: 0
-    //
-    // Depends on: https://github.com/torrust/torrust-tracker/issues/1405
+    if let Some(job) = opt_job {
+        jobs.push(job);
+    }
 }
 
-fn start_udp_core_event_listener(config: &Configuration, app_container: &Arc<AppContainer>) {
-    let _job = jobs::udp_tracker_core::start_event_listener(config, app_container);
+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);
 
-    // todo: the job cannot be added in the jobs vector otherwise the application never ends
-    // because the event listener never stops. You see this console message
-    // forever:
-    //
-    // !! shuting down in 90 seconds !!
-    // 2025-04-24T15:27:45.454101Z  INFO graceful_shutdown: torrust_axum_server::signals: remaining alive connections: 0
-    //
-    // Depends on: https://github.com/torrust/torrust-tracker/issues/1405
+    if let Some(job) = opt_job {
+        jobs.push(job);
+    }
 }
 
-fn start_udp_server_event_listener(config: &Configuration, app_container: &Arc<AppContainer>) {
-    let _job = jobs::udp_tracker_server::start_event_listener(config, app_container);
+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);
 
-    // todo: the job cannot be added in the jobs vector otherwise the application never ends
-    // because the event listener never stops. You see this console message
-    // forever:
-    //
-    // !! shuting down in 90 seconds !!
-    // 2025-04-24T15:27:45.454101Z  INFO graceful_shutdown: torrust_axum_server::signals: remaining alive connections: 0
-    //
-    // Depends on: https://github.com/torrust/torrust-tracker/issues/1405
+    if let Some(job) = opt_job {
+        jobs.push(job);
+    }
 }
 
 async fn start_the_udp_instances(config: &Configuration, app_container: &Arc<AppContainer>, jobs: &mut Vec<JoinHandle<()>>) {

src/bootstrap/jobs/mod.rs

@@ -7,10 +7,10 @@
 //!
 //! This modules contains all the functions needed to start those jobs.
 pub mod health_check_api;
-pub mod http_tracker_core;
 pub mod http_tracker;
+pub mod http_tracker_core;
 pub mod torrent_cleanup;
 pub mod tracker_apis;
+pub mod udp_tracker;
 pub mod udp_tracker_core;
 pub mod udp_tracker_server;
-pub mod udp_tracker;