diff --git a/backend/src/bootstrap/services.rs b/backend/src/bootstrap/services.rs index d73957923..0bb0784a3 100644 --- a/backend/src/bootstrap/services.rs +++ b/backend/src/bootstrap/services.rs @@ -546,6 +546,7 @@ pub async fn run(config: BootstrapConfig) -> Result<(), Box Result<(), Box result, - () = crate::bootstrap::shutdown::shutdown_signal() => { - let _ = graceful_shutdown_tx.send(()); - let active_websocket_connections = connection_drain.begin_shutdown(); - tracing::info!( - active_websocket_connections, - "正在排空入站连接并关闭运行时任务" - ); - let coordinator = task_coordinator.take(); - let connection_drain = connection_drain.clone(); - let runtime_shutdown = async move { - let shutdown_tasks = async move { - if let Some(coordinator) = coordinator { - coordinator.shutdown().await; - } - }; - tokio::join!(shutdown_tasks, connection_drain.wait()); - }; - let drain_runtime = async { - let (serve_result, ()) = tokio::join!(&mut serve, runtime_shutdown); - serve_result - }; - tokio::pin!(drain_runtime); - tokio::select! { - result = &mut drain_runtime => result, - () = tokio::time::sleep(HTTP_DRAIN_TIMEOUT) => { - tracing::warn!( - timeout_secs = HTTP_DRAIN_TIMEOUT.as_secs(), - "HTTP graceful shutdown timed out; cancelling remaining connections" - ); + action = crate::bootstrap::shutdown::shutdown_signal() => { + match action { + crate::bootstrap::shutdown::ShutdownAction::Restart(executable_path) => { + restart_path = Some(executable_path); Ok(()) } - () = crate::bootstrap::shutdown::shutdown_signal() => { - tracing::warn!("Received a second shutdown signal; cancelling remaining connections"); - Ok(()) + crate::bootstrap::shutdown::ShutdownAction::Graceful => { + let _ = graceful_shutdown_tx.send(()); + let active_websocket_connections = connection_drain.begin_shutdown(); + tracing::info!( + active_websocket_connections, + "正在排空入站连接并关闭运行时任务" + ); + let coordinator = task_coordinator.take(); + let connection_drain = connection_drain.clone(); + let runtime_shutdown = async move { + let shutdown_tasks = async move { + if let Some(coordinator) = coordinator { + coordinator.shutdown().await; + } + }; + tokio::join!(shutdown_tasks, connection_drain.wait()); + }; + let drain_runtime = async { + let (serve_result, ()) = tokio::join!(&mut serve, runtime_shutdown); + serve_result + }; + tokio::pin!(drain_runtime); + tokio::select! { + result = &mut drain_runtime => result, + () = tokio::time::sleep(HTTP_DRAIN_TIMEOUT) => { + tracing::warn!( + timeout_secs = HTTP_DRAIN_TIMEOUT.as_secs(), + "HTTP graceful shutdown timed out; cancelling remaining connections" + ); + Ok(()) + } + _ = crate::bootstrap::shutdown::shutdown_signal() => { + tracing::warn!("Received a second shutdown signal; cancelling remaining connections"); + Ok(()) + } + } } } } } }; + if let Some(executable_path) = restart_path { + tracing::info!( + executable = %executable_path.display(), + "正在以更新后的二进制替换当前进程" + ); + drop(log_guard); + return exec_replacement_process(executable_path); + } + if let Some(coordinator) = task_coordinator.take() { let active_websocket_connections = connection_drain.begin_shutdown(); tracing::info!( @@ -610,15 +628,6 @@ pub async fn run(config: BootstrapConfig) -> Result<(), Box> = OnceLock::new(); -static RESTART_REQUESTED: AtomicBool = AtomicBool::new(false); -static RESTART_EXECUTABLE_PATH: OnceLock = OnceLock::new(); +static SHUTDOWN_REQUESTS: OnceLock> = OnceLock::new(); + +#[derive(Clone)] +pub(crate) enum ShutdownAction { + Graceful, + Restart(PathBuf), +} #[derive(Debug, Clone, Copy, Default)] pub struct RuntimeProcessControl; @@ -24,12 +22,12 @@ impl crate::api::router::ProcessControl for RuntimeProcessControl { } fn request_restart(&self, executable_path: PathBuf) { - request_process_restart(executable_path); + request_immediate_restart(executable_path); } } -/// 等待进程关闭信号(Ctrl+C 或 SIGTERM)。 -pub async fn shutdown_signal() { +/// 等待进程关闭或重启信号。 +pub(crate) async fn shutdown_signal() -> ShutdownAction { let mut requested = shutdown_sender().subscribe(); let ctrl_c = async { signal::ctrl_c() @@ -49,33 +47,25 @@ pub async fn shutdown_signal() { let terminate = std::future::pending::<()>(); tokio::select! { - _ = requested.recv() => {}, - () = ctrl_c => {}, - () = terminate => {}, + action = requested.recv() => match action { + Ok(action) => action, + Err(_) => ShutdownAction::Graceful, + }, + () = ctrl_c => ShutdownAction::Graceful, + () = terminate => ShutdownAction::Graceful, } } /// 请求进程按优雅关闭路径退出。 pub fn request_shutdown() { - let _ = shutdown_sender().send(()); + let _ = shutdown_sender().send(ShutdownAction::Graceful); } -/// 请求进程完成优雅关闭后以新二进制替换当前进程。 -pub fn request_process_restart(executable_path: PathBuf) { - let _ = RESTART_EXECUTABLE_PATH.set(executable_path); - RESTART_REQUESTED.store(true, Ordering::SeqCst); - request_shutdown(); +fn request_immediate_restart(executable_path: PathBuf) { + let _ = shutdown_sender().send(ShutdownAction::Restart(executable_path)); } -/// 获取待 exec 的重启目标。 -pub fn restart_executable_path() -> Option { - RESTART_REQUESTED - .load(Ordering::SeqCst) - .then(|| RESTART_EXECUTABLE_PATH.get().cloned()) - .flatten() -} - -fn shutdown_sender() -> &'static broadcast::Sender<()> { +fn shutdown_sender() -> &'static broadcast::Sender { SHUTDOWN_REQUESTS.get_or_init(|| { let (sender, _receiver) = broadcast::channel(16); sender