fix(backend): skip drain for self-restart

Route in-process update restarts through an explicit shutdown action so the runtime releases the listener and logging guard before exec, without waiting for the single instance's graceful drain.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-07-17 18:54:33 +08:00
co-authored by Codex
parent 3c4bd8757f
commit 79c2c5948f
2 changed files with 72 additions and 73 deletions
+50 -41
View File
@@ -546,6 +546,7 @@ pub async fn run(config: BootstrapConfig) -> Result<(), Box<dyn Error + Send + S
let listener = tokio::net::TcpListener::bind((host.as_str(), port)).await?;
let (router, task_coordinator, connection_drain) = build_router(config).await?;
let mut task_coordinator = Some(task_coordinator);
let mut restart_path = None;
let serve_result = {
let (graceful_shutdown_tx, graceful_shutdown_rx) = tokio::sync::oneshot::channel();
@@ -560,46 +561,63 @@ pub async fn run(config: BootstrapConfig) -> Result<(), Box<dyn Error + Send + S
tokio::pin!(serve);
tokio::select! {
result = &mut serve => 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<dyn Error + Send + S
}
serve_result?;
if let Some(executable_path) = crate::bootstrap::shutdown::restart_executable_path() {
tracing::info!(
executable = %executable_path.display(),
"正在以更新后的二进制替换当前进程"
);
drop(log_guard);
exec_replacement_process(executable_path)?;
}
Ok(())
}
+22 -32
View File
@@ -1,19 +1,17 @@
//! 优雅关闭信号处理。
//! 进程关闭与重启信号处理。
use std::{
path::PathBuf,
sync::{
OnceLock,
atomic::{AtomicBool, Ordering},
},
};
use std::{path::PathBuf, sync::OnceLock};
use tokio::signal;
use tokio::sync::broadcast;
static SHUTDOWN_REQUESTS: OnceLock<broadcast::Sender<()>> = OnceLock::new();
static RESTART_REQUESTED: AtomicBool = AtomicBool::new(false);
static RESTART_EXECUTABLE_PATH: OnceLock<PathBuf> = OnceLock::new();
static SHUTDOWN_REQUESTS: OnceLock<broadcast::Sender<ShutdownAction>> = 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<PathBuf> {
RESTART_REQUESTED
.load(Ordering::SeqCst)
.then(|| RESTART_EXECUTABLE_PATH.get().cloned())
.flatten()
}
fn shutdown_sender() -> &'static broadcast::Sender<()> {
fn shutdown_sender() -> &'static broadcast::Sender<ShutdownAction> {
SHUTDOWN_REQUESTS.get_or_init(|| {
let (sender, _receiver) = broadcast::channel(16);
sender