Migrate service manager to trzcina + adopt ServiceBundle

Cargo.toml

@@ -70,6 +70,7 @@ tokio-test = "0.4.4"
 tokio-tungstenite = "0.28"
 tokio-util = "0.7"
 thiserror = "2"
+trzcina = "=0.2.1"
 url = { version = "2.5", features = ["serde"] }
 paddler = { version = "4.0.0", path = "paddler" }
 paddler_bootstrap = { version = "4.0.0", path = "paddler_bootstrap" }

paddler/Cargo.toml

@@ -51,6 +51,7 @@ tokio = { workspace = true }
 tokio-stream = { workspace = true }
 tokio-tungstenite = { workspace = true }
 tokio-util = { workspace = true }
+trzcina = { workspace = true }
 url = { workspace = true }
 
 # web dashboard deps

paddler/src/agent/llamacpp_arbiter_service.rs

@@ -12,6 +12,7 @@ use tokio::time::Duration;
 use tokio::time::MissedTickBehavior;
 use tokio::time::interval;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
 
 use crate::agent::continue_from_conversation_history_request::ContinueFromConversationHistoryRequest;
 use crate::agent::continue_from_raw_prompt_request::ContinueFromRawPromptRequest;
@@ -24,7 +25,6 @@ use crate::agent::generate_embedding_batch_request::GenerateEmbeddingBatchReques
 use crate::agent::model_metadata_holder::ModelMetadataHolder;
 use crate::agent_applicable_state::AgentApplicableState;
 use crate::agent_applicable_state_holder::AgentApplicableStateHolder;
-use crate::service::Service;
 use crate::slot_aggregated_status_manager::SlotAggregatedStatusManager;
 
 pub struct LlamaCppArbiterService {

paddler/src/agent/management_socket_client_service.rs

@@ -1,6 +1,5 @@
 use std::sync::Arc;
 
-use actix_web::rt;
 use actix_web::web::Bytes;
 use anyhow::Context;
 use anyhow::Result;
@@ -18,6 +17,7 @@ use tokio::time::interval;
 use tokio_tungstenite::connect_async;
 use tokio_tungstenite::tungstenite::protocol::Message;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
 
 use paddler_types::agent_desired_state::AgentDesiredState;
 use paddler_types::jsonrpc::Error as JsonRpcError;
@@ -42,7 +42,6 @@ use crate::balancer::management_service::http_route::api::ws_agent_socket::jsonr
 use crate::balancer::management_service::http_route::api::ws_agent_socket::jsonrpc::notification_params::RegisterAgentParams;
 use crate::balancer::management_service::http_route::api::ws_agent_socket::jsonrpc::notification_params::UpdateAgentStatusParams;
 use crate::produces_snapshot::ProducesSnapshot;
-use crate::service::Service;
 use crate::slot_aggregated_status::SlotAggregatedStatus;
 use crate::subscribes_to_updates::SubscribesToUpdates as _;
 
@@ -266,7 +265,7 @@ impl ManagementSocketClientService {
             Message::Text(text) => {
                 let connection_close = incoming_message_context.connection_close.clone();
 
-                rt::spawn(async move {
+                tokio::spawn(async move {
                     tokio::select! {
                         () = connection_close.cancelled() => {
                             info!("Connection close signal received, shutting down");
@@ -327,7 +326,7 @@ impl ManagementSocketClientService {
         let forward_connection_close = connection_close.clone();
         let forward_shutdown = shutdown.clone();
 
-        let message_forward_handle = rt::spawn(async move {
+        let message_forward_handle = tokio::spawn(async move {
             loop {
                 tokio::select! {
                     () = forward_connection_close.cancelled() => {

paddler/src/agent/reconciliation_service.rs

@@ -9,11 +9,11 @@ use tokio::time::Duration;
 use tokio::time::MissedTickBehavior;
 use tokio::time::interval;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
 
 use crate::agent_applicable_state_holder::AgentApplicableStateHolder;
 use crate::agent_issue_fix::AgentIssueFix;
 use crate::converts_to_applicable_state::ConvertsToApplicableState as _;
-use crate::service::Service;
 use crate::slot_aggregated_status::SlotAggregatedStatus;
 
 pub struct ReconciliationService {

paddler/src/balancer/compatibility/openai_service/mod.rs

@@ -10,19 +10,21 @@ use actix_web::web::Data;
 use anyhow::Result;
 use async_trait::async_trait;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
+use trzcina::ServiceShutdownOptions;
 
 use crate::balancer::buffered_request_manager::BufferedRequestManager;
 use crate::balancer::compatibility::openai_service::app_data::AppData;
 use crate::balancer::compatibility::openai_service::configuration::Configuration as OpenAIServiceConfiguration;
 use crate::balancer::http_route as common_http_route;
 use crate::balancer::inference_service::configuration::Configuration as InferenceServiceConfiguration;
 use crate::create_cors_middleware::create_cors_middleware;
-use crate::service::Service;
 
 pub struct OpenAIService {
     pub buffered_request_manager: Arc<BufferedRequestManager>,
     pub inference_service_configuration: InferenceServiceConfiguration,
     pub openai_service_configuration: OpenAIServiceConfiguration,
+    pub shutdown_options: ServiceShutdownOptions,
 }
 
 #[async_trait]
@@ -54,6 +56,7 @@ impl Service for OpenAIService {
         .shutdown_signal(async move {
             shutdown.cancelled().await;
         })
+        .shutdown_timeout(self.shutdown_options.cooperative_deadline.as_secs())
         .disable_signals()
         .bind(self.openai_service_configuration.addr)
         .expect("Unable to bind server to address")

paddler/src/balancer/inference_service/mod.rs

@@ -10,6 +10,8 @@ use actix_web::web::Data;
 use anyhow::Result;
 use async_trait::async_trait;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
+use trzcina::ServiceShutdownOptions;
 
 use crate::balancer::agent_controller_pool::AgentControllerPool;
 use crate::balancer::buffered_request_manager::BufferedRequestManager;
@@ -20,13 +22,13 @@ use crate::balancer::inference_service::configuration::Configuration as Inferenc
 use crate::balancer::web_admin_panel_service::configuration::Configuration as WebAdminPanelServiceConfiguration;
 use crate::balancer_applicable_state_holder::BalancerApplicableStateHolder;
 use crate::create_cors_middleware::create_cors_middleware;
-use crate::service::Service;
 
 pub struct InferenceService {
     pub agent_controller_pool: Arc<AgentControllerPool>,
     pub balancer_applicable_state_holder: Arc<BalancerApplicableStateHolder>,
     pub buffered_request_manager: Arc<BufferedRequestManager>,
     pub configuration: InferenceServiceConfiguration,
+    pub shutdown_options: ServiceShutdownOptions,
     #[cfg(feature = "web_admin_panel")]
     pub web_admin_panel_service_configuration: Option<WebAdminPanelServiceConfiguration>,
 }
@@ -70,6 +72,7 @@ impl Service for InferenceService {
         .shutdown_signal(async move {
             shutdown.cancelled().await;
         })
+        .shutdown_timeout(self.shutdown_options.cooperative_deadline.as_secs())
         .disable_signals()
         .bind(self.configuration.addr)
         .expect("Unable to bind server to address")

paddler/src/balancer/management_service/mod.rs

@@ -10,6 +10,8 @@ use actix_web::web::Data;
 use anyhow::Result;
 use async_trait::async_trait;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
+use trzcina::ServiceShutdownOptions;
 
 use crate::balancer::agent_controller_pool::AgentControllerPool;
 use crate::balancer::buffered_request_manager::BufferedRequestManager;
@@ -25,7 +27,6 @@ use crate::balancer::state_database::StateDatabase;
 use crate::balancer::web_admin_panel_service::configuration::Configuration as WebAdminPanelServiceConfiguration;
 use crate::balancer_applicable_state_holder::BalancerApplicableStateHolder;
 use crate::create_cors_middleware::create_cors_middleware;
-use crate::service::Service;
 
 pub struct ManagementService {
     pub agent_controller_pool: Arc<AgentControllerPool>,
@@ -36,6 +37,7 @@ pub struct ManagementService {
     pub embedding_sender_collection: Arc<EmbeddingSenderCollection>,
     pub generate_tokens_sender_collection: Arc<GenerateTokensSenderCollection>,
     pub model_metadata_sender_collection: Arc<ModelMetadataSenderCollection>,
+    pub shutdown_options: ServiceShutdownOptions,
     pub state_database: Arc<dyn StateDatabase>,
     pub statsd_prefix: String,
     #[cfg(feature = "web_admin_panel")]
@@ -95,6 +97,7 @@ impl Service for ManagementService {
         .shutdown_signal(async move {
             shutdown.cancelled().await;
         })
+        .shutdown_timeout(self.shutdown_options.cooperative_deadline.as_secs())
         .disable_signals()
         .bind(self.configuration.addr)
         .expect("Unable to bind server to address")

paddler/src/balancer/reconciliation_service.rs

@@ -9,11 +9,11 @@ use tokio::time::Duration;
 use tokio::time::MissedTickBehavior;
 use tokio::time::interval;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
 
 use crate::balancer::agent_controller_pool::AgentControllerPool;
 use crate::balancer_applicable_state_holder::BalancerApplicableStateHolder;
 use crate::converts_to_applicable_state::ConvertsToApplicableState as _;
-use crate::service::Service;
 use crate::sets_desired_state::SetsDesiredState as _;
 
 pub struct ReconciliationService {

paddler/src/balancer/statsd_service/mod.rs

@@ -12,12 +12,12 @@ use log::error;
 use tokio::time::MissedTickBehavior;
 use tokio::time::interval;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
 
 use crate::balancer::agent_controller_pool::AgentControllerPool;
 use crate::balancer::agent_controller_pool_total_slots::AgentControllerPoolTotalSlots;
 use crate::balancer::buffered_request_manager::BufferedRequestManager;
 use crate::balancer::statsd_service::configuration::Configuration as StatsdServiceConfiguration;
-use crate::service::Service;
 
 pub struct StatsdService {
     pub agent_controller_pool: Arc<AgentControllerPool>,

paddler/src/balancer/web_admin_panel_service/mod.rs

@@ -9,13 +9,15 @@ use actix_web::web::Data;
 use anyhow::Result;
 use async_trait::async_trait;
 use tokio_util::sync::CancellationToken;
+use trzcina::Service;
+use trzcina::ServiceShutdownOptions;
 
 use crate::balancer::web_admin_panel_service::app_data::AppData;
 use crate::balancer::web_admin_panel_service::configuration::Configuration as WebAdminPanelServiceConfiguration;
-use crate::service::Service;
 
 pub struct WebAdminPanelService {
     pub configuration: WebAdminPanelServiceConfiguration,
+    pub shutdown_options: ServiceShutdownOptions,
 }
 
 #[async_trait]
@@ -40,6 +42,7 @@ impl Service for WebAdminPanelService {
         .shutdown_signal(async move {
             shutdown.cancelled().await;
         })
+        .shutdown_timeout(self.shutdown_options.cooperative_deadline.as_secs())
         .disable_signals()
         .bind(self.configuration.addr)
         .expect("Unable to bind server to address")

paddler/src/lib.rs

@@ -29,8 +29,6 @@ pub mod resolve_desired_model;
 pub mod resolved_socket_addr;
 pub mod resolves_model_source;
 pub mod sends_rpc_message;
-pub mod service;
-pub mod service_manager;
 pub mod sets_desired_state;
 pub mod slot_aggregated_status;
 pub mod slot_aggregated_status_download_progress;