Merge #1487: Fix shutdown message and improve it

8f42271 chore: [#1477] remove unused dependency (Jose Celano)
53fdafd feat: [#1477] extract JobManager to handle jobs (Jose Celano)
faf8111 fix: [#1477] shutdown event listeners on CRTL+c signal (Jose Celano)
e595190 refactor: [#1477] create an explicit job creation module for event listeners (Jose Celano)
a29aaae feat: [#1477] add server address to http server shutdown logs (Jose Celano)
bbc0154 feat: [#1477] add fallback timeout to graceful shutdown (Jose Celano)
797583b feat: [#1477] shut down http server inmediately is all coneections closed (Jose Celano)
d5e7c36 feat: [#1477] improve shutting down message for http servers (Jose Celano)

Pull request description:

  Fix shutdown message and improve it for HTTP servers (they all use Axum).

  ### Subtasks

  - [x] Hard timeout for shutdown. Stop the loop even when there are alive connections after the grace period.
  - [x] Show socket-bound address in logs.
  - [x] Fix bug, not showing the message passed as a parameter.
  - [x] Event listeners should listen to `CRTL+c` signal. Temporary solution until we implement:
    - #1405.
  - [x] Add a job manager to give a name to all jobs and identify them when the app shuts down.

ACKs for top commit:
  josecelano:
    ACK 8f42271

Tree-SHA512: 5a8b726e5eb6362020e46c82a7dc85420c39d3e7b072f46ecf99fc282179f53e64a7bf979a055d351a7f76f65a1681572d657dbcc5bb42ba2409b77d230696dd

Author: josecelano

Cargo.lock

@@ -40,7 +40,6 @@ bittorrent-tracker-core = { version = "3.0.0-develop", path = "packages/tracker-
 bittorrent-udp-tracker-core = { version = "3.0.0-develop", path = "packages/udp-tracker-core" }
 chrono = { version = "0", default-features = false, features = ["clock"] }
 clap = { version = "4", features = ["derive", "env"] }
-futures = "0"
 rand = "0"
 regex = "1"
 reqwest = { version = "0", features = ["json"] }

Cargo.toml

@@ -113,6 +113,7 @@ pub fn start(
         handle.clone(),
         rx_halt,
         format!("Shutting down http server on socket address: {address}"),
+        address,
     ));
 
     let running = axum_server::from_tcp(socket)

packages/axum-health-check-api-server/src/server.rs

@@ -60,6 +60,7 @@ impl Launcher {
             handle.clone(),
             rx_halt,
             format!("Shutting down HTTP server on socket address: {address}"),
+            address,
         ));
 
         let tls = self.tls.clone();

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

@@ -257,6 +257,7 @@ impl Launcher {
             handle.clone(),
             rx_halt,
             format!("Shutting down tracker API server on socket address: {address}"),
+            address,
         ));
 
         let tls = self.tls.clone();

packages/axum-rest-tracker-api-server/src/server.rs

@@ -1,21 +1,49 @@
+use std::net::SocketAddr;
 use std::time::Duration;
 
-use tokio::time::sleep;
+use tokio::time::{sleep, Instant};
 use torrust_server_lib::signals::{shutdown_signal_with_message, Halted};
 use tracing::instrument;
 
 #[instrument(skip(handle, rx_halt, message))]
-pub async fn graceful_shutdown(handle: axum_server::Handle, rx_halt: tokio::sync::oneshot::Receiver<Halted>, message: String) {
-    shutdown_signal_with_message(rx_halt, message).await;
+pub async fn graceful_shutdown(
+    handle: axum_server::Handle,
+    rx_halt: tokio::sync::oneshot::Receiver<Halted>,
+    message: String,
+    address: SocketAddr,
+) {
+    shutdown_signal_with_message(rx_halt, message.clone()).await;
 
-    tracing::debug!("Sending graceful shutdown signal");
-    handle.graceful_shutdown(Some(Duration::from_secs(90)));
+    let grace_period = Duration::from_secs(90);
+    let max_wait = Duration::from_secs(95);
+    let start = Instant::now();
 
-    println!("!! shuting down in 90 seconds !!");
+    handle.graceful_shutdown(Some(grace_period));
+
+    tracing::info!("!! {} in {} seconds !!", message, grace_period.as_secs());
 
     loop {
-        sleep(Duration::from_secs(1)).await;
+        if handle.connection_count() == 0 {
+            tracing::info!("All connections closed, shutting down server in address {}", address);
+            break;
+        }
+
+        if start.elapsed() >= max_wait {
+            tracing::warn!(
+                "Shutdown timeout of {} seconds reached. Forcing shutdown in address {} with {} active connections.",
+                max_wait.as_secs(),
+                address,
+                handle.connection_count()
+            );
+            break;
+        }
 
-        tracing::info!("remaining alive connections: {}", handle.connection_count());
+        tracing::info!(
+            "Remaining alive connections: {} ({}s elapsed)",
+            handle.connection_count(),
+            start.elapsed().as_secs()
+        );
+
+        sleep(Duration::from_secs(1)).await;
     }
 }

packages/axum-server/src/signals.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/http-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-core/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);
+                            }
+                        }
                     }
                 }
             }