feat(launcher): proper servers shutdown

libs/blockscout-service-launcher/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "blockscout-service-launcher"
-version = "0.14.0"
+version = "0.15.0"
 description = "Allows to launch blazingly fast blockscout rust services"
 license = "MIT"
 repository = "https://github.com/blockscout/blockscout-rs"
@@ -27,6 +27,7 @@ reqwest = { version = "0.11", features = ["json"], optional = true }
 serde = { version = "1.0", features = ["derive"], optional = true }
 serde_json = {version = "1", optional = true }
 tokio = { version = "1", optional = true }
+tokio-util = { version = "0.7.12", optional = true }
 tonic = { version = "0.8", optional = true }
 tracing = { version = "0.1", optional = true }
 tracing-actix-web = { package = "blockscout-tracing-actix-web", version = "0.8.0", optional = true }
@@ -68,6 +69,7 @@ launcher = [
     "dep:prometheus",
     "dep:serde",
     "dep:tokio",
+    "dep:tokio-util",
     "dep:tonic",
     "dep:tracing",
     "dep:tracing-actix-web",

libs/blockscout-service-launcher/src/launcher/launch.rs

@@ -8,8 +8,11 @@ use super::{
 use actix_web::{middleware::Condition, App, HttpServer};
 use actix_web_prom::PrometheusMetrics;
 use std::net::SocketAddr;
+use tokio_util::sync::CancellationToken;
 use tracing_actix_web::TracingLogger;
 
+pub(crate) const SHUTDOWN_TIMEOUT_SEC: u64 = 10;
+
 pub struct LaunchSettings {
     pub service_name: String,
     pub server: ServerSettings,
@@ -20,6 +23,7 @@ pub async fn launch<R>(
     settings: &LaunchSettings,
     http: R,
     grpc: tonic::transport::server::Router,
+    shutdown: Option<CancellationToken>,
 ) -> Result<(), anyhow::Error>
 where
     R: HttpRouter + Send + Sync + Clone + 'static,
@@ -39,6 +43,7 @@ where
                     .as_ref()
                     .map(|metrics| metrics.http_middleware().clone()),
                 &settings.server.http,
+                shutdown.clone(),
             );
             tokio::spawn(async move { http_server_future.await.map_err(anyhow::Error::msg) })
         };
@@ -47,7 +52,7 @@ where
 
     if settings.server.grpc.enabled {
         let grpc_server = {
-            let grpc_server_future = grpc_serve(grpc, settings.server.grpc.addr);
+            let grpc_server_future = grpc_serve(grpc, settings.server.grpc.addr, shutdown.clone());
             tokio::spawn(async move { grpc_server_future.await.map_err(anyhow::Error::msg) })
         };
         futures.push(grpc_server)
@@ -56,7 +61,7 @@ where
     if let Some(metrics) = metrics {
         let addr = settings.metrics.addr;
         futures.push(tokio::spawn(async move {
-            metrics.run_server(addr).await?;
+            metrics.run_server(addr, shutdown).await?;
             Ok(())
         }));
     }
@@ -68,10 +73,29 @@ where
     res?
 }
 
+pub(crate) async fn stop_actix_server_on_cancel(
+    actix_handle: actix_web::dev::ServerHandle,
+    shutdown: CancellationToken,
+    graceful: bool,
+) {
+    shutdown.cancelled().await;
+    tracing::info!(
+        "Shutting down actix server (gracefully: {graceful}).\
+        Should finish within {SHUTDOWN_TIMEOUT_SEC} seconds..."
+    );
+    actix_handle.stop(graceful).await;
+}
+
+pub(crate) async fn grpc_cancel_signal(shutdown: CancellationToken) {
+    shutdown.cancelled().await;
+    tracing::info!("Shutting down grpc server...");
+}
+
 fn http_serve<R>(
     http: R,
     metrics: Option<PrometheusMetrics>,
     settings: &HttpServerSettings,
+    shutdown: Option<CancellationToken>,
 ) -> actix_web::dev::Server
 where
     R: HttpRouter + Send + Sync + Clone + 'static,
@@ -84,7 +108,7 @@ where
     let json_cfg = actix_web::web::JsonConfig::default().limit(settings.max_body_size);
     let cors_settings = settings.cors.clone();
     let cors_enabled = cors_settings.enabled;
-    if let Some(metrics) = metrics {
+    let server = if let Some(metrics) = metrics {
         HttpServer::new(move || {
             let cors = cors_settings.clone().build();
             App::new()
@@ -94,6 +118,7 @@ where
                 .app_data(json_cfg.clone())
                 .configure(configure_router(&http))
         })
+        .shutdown_timeout(SHUTDOWN_TIMEOUT_SEC)
         .bind(settings.addr)
         .expect("failed to bind server")
         .run()
@@ -106,16 +131,27 @@ where
                 .app_data(json_cfg.clone())
                 .configure(configure_router(&http))
         })
+        .shutdown_timeout(SHUTDOWN_TIMEOUT_SEC)
         .bind(settings.addr)
         .expect("failed to bind server")
         .run()
+    };
+    if let Some(shutdown) = shutdown {
+        tokio::spawn(stop_actix_server_on_cancel(server.handle(), shutdown, true));
     }
+    server
 }
 
-fn grpc_serve(
+async fn grpc_serve(
     grpc: tonic::transport::server::Router,
     addr: SocketAddr,
-) -> impl futures::Future<Output = Result<(), tonic::transport::Error>> {
+    shutdown: Option<CancellationToken>,
+) -> Result<(), tonic::transport::Error> {
     tracing::info!("starting grpc server on addr {}", addr);
-    grpc.serve(addr)
+    if let Some(shutdown) = shutdown {
+        grpc.serve_with_shutdown(addr, grpc_cancel_signal(shutdown))
+            .await
+    } else {
+        grpc.serve(addr).await
+    }
 }

libs/blockscout-service-launcher/src/launcher/metrics.rs

@@ -1,6 +1,9 @@
 use actix_web::{App, HttpServer};
 use actix_web_prom::{PrometheusMetrics, PrometheusMetricsBuilder};
 use std::{collections::HashMap, net::SocketAddr};
+use tokio_util::sync::CancellationToken;
+
+use crate::launcher::launch::{stop_actix_server_on_cancel, SHUTDOWN_TIMEOUT_SEC};
 
 #[derive(Clone)]
 pub struct Metrics {
@@ -33,11 +36,20 @@ impl Metrics {
         &self.http_middleware
     }
 
-    pub fn run_server(self, addr: SocketAddr) -> actix_web::dev::Server {
+    pub fn run_server(
+        self,
+        addr: SocketAddr,
+        shutdown: Option<CancellationToken>,
+    ) -> actix_web::dev::Server {
         tracing::info!(addr = ?addr, "starting metrics server");
-        HttpServer::new(move || App::new().wrap(self.metrics_middleware.clone()))
+        let server = HttpServer::new(move || App::new().wrap(self.metrics_middleware.clone()))
+            .shutdown_timeout(SHUTDOWN_TIMEOUT_SEC)
             .bind(addr)
             .unwrap()
-            .run()
+            .run();
+        if let Some(shutdown) = shutdown {
+            tokio::spawn(stop_actix_server_on_cancel(server.handle(), shutdown, true));
+        }
+        server
     }
 }