feat: server watchdog 子命令,崩溃自动重启加信号优雅停栈

This commit is contained in:
2026-10-06 19:29:24 +00:00
parent e199ebeaa1
commit 2a4da64613
7 changed files with 668 additions and 90 deletions
Generated
+18 -18
View File
@@ -1160,7 +1160,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_builtins" name = "oxide_builtins"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"chrono", "chrono",
"fast-float", "fast-float",
@@ -1185,7 +1185,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_bytecode" name = "oxide_bytecode"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"num-bigint", "num-bigint",
"oxide_log", "oxide_log",
@@ -1193,7 +1193,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_cfg" name = "oxide_cfg"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"oxide_bytecode", "oxide_bytecode",
"oxide_emit", "oxide_emit",
@@ -1204,7 +1204,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_cli" name = "oxide_cli"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"ansi_term", "ansi_term",
"bumpalo", "bumpalo",
@@ -1225,7 +1225,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_code_cache" name = "oxide_code_cache"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"lru", "lru",
"oxide_bytecode", "oxide_bytecode",
@@ -1234,7 +1234,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_compiler" name = "oxide_compiler"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"oxide_bytecode", "oxide_bytecode",
"oxide_cfg", "oxide_cfg",
@@ -1253,7 +1253,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_dce" name = "oxide_dce"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"oxide_bytecode", "oxide_bytecode",
"oxide_cfg", "oxide_cfg",
@@ -1267,7 +1267,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_emit" name = "oxide_emit"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"num-bigint", "num-bigint",
"oxc_allocator", "oxc_allocator",
@@ -1282,7 +1282,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_ir" name = "oxide_ir"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"oxide_bytecode", "oxide_bytecode",
"oxide_log", "oxide_log",
@@ -1291,7 +1291,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_kernel" name = "oxide_kernel"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"bumpalo", "bumpalo",
"dashmap", "dashmap",
@@ -1306,7 +1306,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_liveness" name = "oxide_liveness"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"oxide_bytecode", "oxide_bytecode",
"oxide_cfg", "oxide_cfg",
@@ -1318,7 +1318,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_log" name = "oxide_log"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"tracing", "tracing",
"tracing-appender", "tracing-appender",
@@ -1327,7 +1327,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_parser" name = "oxide_parser"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"oxc_allocator", "oxc_allocator",
"oxc_ast", "oxc_ast",
@@ -1342,7 +1342,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_regalloc" name = "oxide_regalloc"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"oxide_bytecode", "oxide_bytecode",
"oxide_cfg", "oxide_cfg",
@@ -1358,7 +1358,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_runtime_api" name = "oxide_runtime_api"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"num-bigint", "num-bigint",
"num-traits", "num-traits",
@@ -1370,7 +1370,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_test262" name = "oxide_test262"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"oxide_bytecode", "oxide_bytecode",
"oxide_compiler", "oxide_compiler",
@@ -1387,7 +1387,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_types" name = "oxide_types"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"bumpalo", "bumpalo",
"hashbrown 0.14.5", "hashbrown 0.14.5",
@@ -1399,7 +1399,7 @@ dependencies = [
[[package]] [[package]]
name = "oxide_vm" name = "oxide_vm"
version = "0.8.41" version = "0.8.42"
dependencies = [ dependencies = [
"bumpalo", "bumpalo",
"chrono", "chrono",
+1 -1
View File
@@ -22,7 +22,7 @@ members = [
] ]
[workspace.package] [workspace.package]
version = "0.8.41" version = "0.8.42"
edition = "2021" edition = "2021"
license = "MIT" license = "MIT"
rust-version = "1.80" rust-version = "1.80"
+19 -63
View File
@@ -2,7 +2,7 @@
#![allow(dead_code)] #![allow(dead_code)]
use std::fs; use std::fs;
use std::process::{Command, ExitCode, Stdio}; use std::process::ExitCode;
use std::sync::Arc; use std::sync::Arc;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
@@ -15,6 +15,8 @@ use oxide_cli::server::log as server_log;
use oxide_cli::server::protocol::{ForgeTarget, LookupResult, ServerRequest, ServerResponse}; use oxide_cli::server::protocol::{ForgeTarget, LookupResult, ServerRequest, ServerResponse};
use oxide_cli::server::server::{run_server, run_server_rm, RmServerConfig, ServerConfig}; use oxide_cli::server::server::{run_server, run_server_rm, RmServerConfig, ServerConfig};
use oxide_cli::server::sidecar; use oxide_cli::server::sidecar;
use oxide_cli::server::spawn;
use oxide_cli::server::watchdog;
use oxide_compiler::compiler::{compiled_module_hash, Compiler}; use oxide_compiler::compiler::{compiled_module_hash, Compiler};
use oxide_compiler::compiler_error; use oxide_compiler::compiler_error;
use oxide_kernel::kernel::{KernelConfig, KernelCore}; use oxide_kernel::kernel::{KernelConfig, KernelCore};
@@ -169,6 +171,12 @@ enum ServerCommands {
#[arg(long)] #[arg(long)]
lookup: Option<String>, lookup: Option<String>,
}, },
/// 监控 server 存活并在崩溃后自动重启(前台进程,60 秒窗口 5 次崩溃即停)。
Watchdog {
/// 常驻 worker 线程数;缺省取宿主核数。
#[arg(long)]
workers: Option<u32>,
},
} }
/// forge 查询目标(CLI 侧,映射到协议侧 `ForgeTarget`)。 /// forge 查询目标(CLI 侧,映射到协议侧 `ForgeTarget`)。
@@ -281,10 +289,8 @@ fn main() -> ExitCode {
} }
} }
/// server 子命令分派:start(两形态)与 cleanup 为最小实现, /// server 子命令分派:全部十一臂为真实实现(start 两形态、cleanup、
/// version/status/health/info/stop/restart 六臂为真实实现, /// 五个控制臂、restart、log、forge、watchdog)。
/// log 为真实实现(读 well-known 日志文件,支持 --follow/--level/--lines),
/// forge 为真实实现(查询四张共享 forge 状态,加三行动旗标)。
/// ///
/// # 步骤 /// # 步骤
/// 1. Start:rm 为真调独立模式入口(进程唯一路径);foreground 为真当前进程 /// 1. Start:rm 为真调独立模式入口(进程唯一路径);foreground 为真当前进程
@@ -301,6 +307,8 @@ fn main() -> ExitCode {
/// `--follow` 时进入阻塞跟踪循环(详见 log_command)。 /// `--follow` 时进入阻塞跟踪循环(详见 log_command)。
/// 6. Forge:发 forge 查询请求(目标加三行动旗标),成功打印渲染文本退 0, /// 6. Forge:发 forge 查询请求(目标加三行动旗标),成功打印渲染文本退 0,
/// 失败打印错误消息退 1。 /// 失败打印错误消息退 1。
/// 7. Watchdog:前台监控进程,拉起 server、三态裁决、崩溃后自动重启
/// (详见 run_watchdog)。
/// ///
/// # 边界与前提 /// # 边界与前提
/// - `--idle-timeout` 不带 --rm 时静默忽略(语义门控归后续任务)。 /// - `--idle-timeout` 不带 --rm 时静默忽略(语义门控归后续任务)。
@@ -311,6 +319,7 @@ fn main() -> ExitCode {
/// 收领继续运行)。 /// 收领继续运行)。
/// - cleanup 可能向 sidecar 记录的进程发 SIGTERM 并删除残留文件。 /// - cleanup 可能向 sidecar 记录的进程发 SIGTERM 并删除残留文件。
/// - 控制臂向 server 建立并关闭一条 Unix socket 连接。 /// - 控制臂向 server 建立并关闭一条 Unix socket 连接。
/// - watchdog spawn 脱离的 server 子进程,崩溃后经同一入口重启。
fn server_command(command: ServerCommands) -> ExitCode { fn server_command(command: ServerCommands) -> ExitCode {
match command { match command {
ServerCommands::Start { rm, idle_timeout, workers, foreground } => { ServerCommands::Start { rm, idle_timeout, workers, foreground } => {
@@ -411,6 +420,7 @@ fn server_command(command: ServerCommands) -> ExitCode {
_ => None, _ => None,
}) })
} }
ServerCommands::Watchdog { workers } => watchdog::run_watchdog(workers),
} }
} }
@@ -484,13 +494,13 @@ fn control_arm(request: &ServerRequest, render: fn(&ServerResponse) -> Option<St
/// - server 为脱离进程,CLI 退出后由 init 收领继续运行。 /// - server 为脱离进程,CLI 退出后由 init 收领继续运行。
fn start_daemon(workers: Option<u32>) -> ExitCode { fn start_daemon(workers: Option<u32>) -> ExitCode {
// spawn:脱离子进程,标准流全空,不等待。 // spawn:脱离子进程,标准流全空,不等待。
if let Err(err) = spawn_detached_server(workers) { if let Err(err) = spawn::spawn_detached_server(workers) {
eprintln!("{}", Red.paint(format!("spawn server 进程失败:{err}"))); eprintln!("{}", Red.paint(format!("spawn server 进程失败:{err}")));
return ExitCode::FAILURE; return ExitCode::FAILURE;
} }
// 就绪探测:单循环全覆盖「sidecar 出现 + 健康请求通过」。 // 就绪探测:单循环全覆盖「sidecar 出现 + 健康请求通过」。
if let Err(msg) = wait_server_ready(Duration::from_secs(10)) { if let Err(msg) = spawn::wait_server_ready(Duration::from_secs(10)) {
eprintln!("{}", Red.paint(format!("{msg}(可用 `oxide server cleanup` 人工检查)"))); eprintln!("{}", Red.paint(format!("{msg}(可用 `oxide server cleanup` 人工检查)")));
return ExitCode::FAILURE; return ExitCode::FAILURE;
} }
@@ -499,60 +509,6 @@ fn start_daemon(workers: Option<u32>) -> ExitCode {
ExitCode::SUCCESS ExitCode::SUCCESS
} }
/// spawn 脱离的前台 server 子进程(`server start --foreground`):标准流全空,不等待。
///
/// # 步骤
/// 1. 取当前可执行文件路径。
/// 2. 构造命令 `server start --foreground` 加可选 `--workers N`。
/// 3. 三个标准流置空并 spawn。
///
/// # 边界与前提
/// - 必须带 `--foreground`:守护形态 spawn 的是前台 server 入口,不带该旗标会
/// 再次进入守护形态形成递归 spawn。
///
/// # 副作用
/// - 创建脱离的子进程,调用方退出后由 init 收领继续运行。
fn spawn_detached_server(workers: Option<u32>) -> std::io::Result<std::process::Child> {
let exe = std::env::current_exe()?;
let mut cmd = Command::new(exe);
cmd.args(["server", "start", "--foreground"]);
if let Some(n) = workers {
cmd.args(["--workers", &n.to_string()]);
}
cmd.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
}
/// 就绪探测:轮询健康请求直至得 healthy。
///
/// # 步骤
/// 1. 循环发健康请求,得 healthy 即返回成功。
/// 2. 超过截止返回含超时秒数的消息。
///
/// # 边界与前提
/// - 控制客户端发请求前先读 sidecar 取 socket 路径,sidecar 缺失时自然走
/// 连接失败分支,「sidecar 出现」与「健康请求通过」一个循环全覆盖
/// (socket 文件出现蕴含监听器已绑定,健康请求通过蕴含 accept 循环在运行)。
///
/// # 副作用
/// - 每轮建立并关闭一条 Unix socket 连接。
fn wait_server_ready(deadline: Duration) -> Result<(), String> {
let end = Instant::now() + deadline;
loop {
let ready =
matches!(send_control_request(&ServerRequest::Health), Ok(ServerResponse::Health { healthy: true }));
if ready {
return Ok(());
}
if Instant::now() >= end {
return Err(format!("server 未在 {} 秒内就绪", deadline.as_secs()));
}
std::thread::sleep(Duration::from_millis(100));
}
}
/// restart 子命令:向存活 server 发关闭请求,等其退出后拉起新 server 进程。 /// restart 子命令:向存活 server 发关闭请求,等其退出后拉起新 server 进程。
/// ///
/// # 步骤 /// # 步骤
@@ -591,13 +547,13 @@ fn restart_command() -> ExitCode {
} }
// 拉起新 server:脱离进程,标准流全空,不等待。 // 拉起新 server:脱离进程,标准流全空,不等待。
if let Err(err) = spawn_detached_server(None) { if let Err(err) = spawn::spawn_detached_server(None) {
eprintln!("{}", Red.paint(format!("拉起新 server 失败:{err}"))); eprintln!("{}", Red.paint(format!("拉起新 server 失败:{err}")));
return ExitCode::FAILURE; return ExitCode::FAILURE;
} }
// 就绪探测:健康请求通过即新 server 的监听与 accept 循环在运行。 // 就绪探测:健康请求通过即新 server 的监听与 accept 循环在运行。
if let Err(msg) = wait_server_ready(Duration::from_secs(10)) { if let Err(msg) = spawn::wait_server_ready(Duration::from_secs(10)) {
eprintln!("{}", Red.paint(format!("新 {msg}"))); eprintln!("{}", Red.paint(format!("新 {msg}")));
return ExitCode::FAILURE; return ExitCode::FAILURE;
} }
+8 -2
View File
@@ -2,14 +2,16 @@
//! //!
//! 持久 server 是常驻进程,复用预热池、近零 spawn 成本; //! 持久 server 是常驻进程,复用预热池、近零 spawn 成本;
//! 通信走 Unix socket + NDJSON(换行分隔 JSON)。 //! 通信走 Unix socket + NDJSON(换行分隔 JSON)。
//! 模块分八层:协议数据面(帧格式与 serde 结构体)、身份注册(sidecar 文件 //! 模块分十层:协议数据面(帧格式与 serde 结构体)、身份注册(sidecar 文件
//! 与 liveness 检查原语)、控制客户端(读 sidecar、连接 server、发控制请求 //! 与 liveness 检查原语)、控制客户端(读 sidecar、连接 server、发控制请求
//! 帧、读回响应帧、三分支错误)、执行请求路径(parse → compile → spawn → //! 帧、读回响应帧、三分支错误)、执行请求路径(parse → compile → spawn →
//! run → format → drop 六段纯函数)、worker 池(固定 N 常驻 worker、各持 //! run → format → drop 六段纯函数)、worker 池(固定 N 常驻 worker、各持
//! 自有 VM 池、mpsc 轮询路由)、server 进程主体(accept 循环、控制请求直接 //! 自有 VM 池、mpsc 轮询路由)、server 进程主体(accept 循环、控制请求直接
//! 处理、执行请求路由、优雅退出)、日志读取(行级别解析、阈值加最后 N 行 //! 处理、执行请求路由、优雅退出)、日志读取(行级别解析、阈值加最后 N 行
//! 过滤、阻塞式跟踪)、forge 状态查询(读四张共享 forge 条目数与容量、执行 //! 过滤、阻塞式跟踪)、forge 状态查询(读四张共享 forge 条目数与容量、执行
//! gc / clear-cache / lookup 三旗标)。 //! gc / clear-cache / lookup 三旗标)、spawn 与就绪探测(start、restart、
//! watchdog 三条路径共用的唯一共享入口)、watchdog(前台监控进程、三态裁决
//! 加崩溃预算、崩溃后自动重启)。
/// 人工兜底清理命令:扫 sidecar、探活、杀 PID、删孤儿文件。 /// 人工兜底清理命令:扫 sidecar、探活、杀 PID、删孤儿文件。
pub mod cleanup; pub mod cleanup;
@@ -34,5 +36,9 @@ pub mod protocol;
pub mod server; pub mod server;
/// 身份注册:sidecar 文件、liveness 检查与陈旧清理。 /// 身份注册:sidecar 文件、liveness 检查与陈旧清理。
pub mod sidecar; pub mod sidecar;
/// spawn 与就绪探测:start、restart、watchdog 三条路径共用的唯一共享入口。
pub mod spawn;
/// worker 池:固定 N 常驻 worker、各持自有 VM 池、mpsc 轮询路由。 /// worker 池:固定 N 常驻 worker、各持自有 VM 池、mpsc 轮询路由。
pub mod workers; pub mod workers;
/// watchdog:前台监控进程,三态裁决加崩溃预算,崩溃后自动重启。
pub mod watchdog;
+65
View File
@@ -0,0 +1,65 @@
//! spawn 与就绪探测:拉起脱离 server 与就绪探测的唯一共享入口。
//!
//! start、restart、watchdog 三条路径共用本模块的同一对函数,不另起第二套
//! spawn 逻辑。`current_exe()` 在 lib 上下文与二进制目标返回同一路径(lib
//! 链接进同一个 oxide 二进制),spawn 行为逐字节不变。
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
use super::client::send_control_request;
use super::protocol::{ServerRequest, ServerResponse};
/// spawn 脱离的前台 server 子进程(`server start --foreground`):标准流全空,不等待。
///
/// # 步骤
/// 1. 取当前可执行文件路径。
/// 2. 构造命令 `server start --foreground` 加可选 `--workers N`。
/// 3. 三个标准流置空并 spawn。
///
/// # 边界与前提
/// - 必须带 `--foreground`:守护形态 spawn 的是前台 server 入口,不带该旗标会
/// 再次进入守护形态形成递归 spawn。
///
/// # 副作用
/// - 创建脱离的子进程,调用方退出后由 init 收领继续运行。
pub fn spawn_detached_server(workers: Option<u32>) -> std::io::Result<std::process::Child> {
let exe = std::env::current_exe()?;
let mut cmd = Command::new(exe);
cmd.args(["server", "start", "--foreground"]);
if let Some(n) = workers {
cmd.args(["--workers", &n.to_string()]);
}
cmd.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
}
/// 就绪探测:轮询健康请求直至得 healthy。
///
/// # 步骤
/// 1. 循环发健康请求,得 healthy 即返回成功。
/// 2. 超过截止返回含超时秒数的消息。
///
/// # 边界与前提
/// - 控制客户端发请求前先读 sidecar 取 socket 路径,sidecar 缺失时自然走
/// 连接失败分支,「sidecar 出现」与「健康请求通过」一个循环全覆盖
/// (socket 文件出现蕴含监听器已绑定,健康请求通过蕴含 accept 循环在运行)。
///
/// # 副作用
/// - 每轮建立并关闭一条 Unix socket 连接。
pub fn wait_server_ready(deadline: Duration) -> Result<(), String> {
let end = Instant::now() + deadline;
loop {
let ready =
matches!(send_control_request(&ServerRequest::Health), Ok(ServerResponse::Health { healthy: true }));
if ready {
return Ok(());
}
if Instant::now() >= end {
return Err(format!("server 未在 {} 秒内就绪", deadline.as_secs()));
}
std::thread::sleep(Duration::from_millis(100));
}
}
+374
View File
@@ -0,0 +1,374 @@
//! watchdog:持久 server 的前台监控进程,崩溃后自动重启。
//!
//! 关键约定:
//! - release 模式 `panic=abort` 使 `catch_unwind` 失效,server 崩溃后进程直接
//! 消亡、文件残留,watchdog 是崩溃后恢复的唯一自动手段。
//! - 裁决信号取自既有工件加子进程收领:子进程是监控对象时,其退出状态是
//! 死亡的权威信号(被杀的子进程是僵尸,PID 探活对僵尸误判活);子进程在
//! 认领阶段被拒时监控对象是既有 server,按 sidecar 存在性、PID 与启动时
//! 刻度(liveness 原语,排 PID 复用误判)三信号裁决;日志文件(跨重启的
//! 诊断工件)供崩溃诊断打印。
//! - watchdog 自身不注册身份、不写任何文件;watchdog 消亡后无人自动恢复
//! (纯标准库轮询进程,不建内核、不建池、不执行引擎代码,崩溃风险极低;
//! server 是脱离进程,watchdog 退出或消亡都不影响 server 运行),人工重跑
//! `oxide server watchdog` 即接管(就绪探测看到存活 server 即接管监控)。
//! - SIGINT/SIGTERM 时先尽力向 server 发关闭请求(优雅退出、删文件)再退 0
//! ——停 watchdog 的语义是「停掉整个监控栈」,不留下无人监控的 server。
use std::path::Path;
use std::process::{ExitCode, ExitStatus};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use super::client::send_control_request;
use super::log as server_log;
use super::protocol::ServerRequest;
use super::sidecar::{self, ServerIdentity};
use super::spawn::{spawn_detached_server, wait_server_ready};
/// 监控循环轮询间隔:500 毫秒(与日志跟踪的轮询间隔同值)。
const POLL_INTERVAL: Duration = Duration::from_millis(500);
/// 崩溃重启退避时长:1 秒。
const CRASH_BACKOFF: Duration = Duration::from_secs(1);
/// 崩溃预算窗口:60 秒。
const CRASH_WINDOW: Duration = Duration::from_secs(60);
/// 崩溃预算上限:窗口内 5 次崩溃。
const CRASH_LIMIT: usize = 5;
/// 崩溃诊断打印的日志尾部行数:20 行。
const CRASH_LOG_LINES: usize = 20;
/// 就绪探测截止:10 秒(与 start / restart 路径同值)。
const READY_DEADLINE: Duration = Duration::from_secs(10);
/// server 状态三态:存活、崩溃、已退出。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WatchState {
/// 存活:PID 活且启动时刻刻度与 sidecar 记录匹配。
Alive,
/// 崩溃:sidecar 残留且 PID 死,或 PID 活但刻度失配(PID 已被复用)。
Crash,
/// 已退出:sidecar 消失(优雅退出序列的最后删除项,或外部清理)。
Exited,
}
/// 裁决表:sidecar 可读性、PID 存活、刻度匹配三信号映射三态。
///
/// # 边界与前提
/// - 纯函数无 I/O,信号由调用方采集(与 cleanup 的 decide 同型)。
pub fn decide(identity: Option<&ServerIdentity>, pid_alive: bool, starttime_match: bool) -> WatchState {
match identity {
None => WatchState::Exited,
Some(_) if !pid_alive => WatchState::Crash,
Some(_) if starttime_match => WatchState::Alive,
Some(_) => WatchState::Crash,
}
}
/// 崩溃预算:`window` 滑动窗口内至多 `limit` 次崩溃,超出即放弃自动重启。
///
/// 窗口语义:窗口内崩溃次数超过上限即判 server 不稳定,继续重启只会空转
/// (典型场景:二进制损坏导致每次启动即崩溃,但每次都通过了就绪探测)。
#[derive(Debug)]
pub struct CrashBudget {
window: Duration,
limit: usize,
times: Vec<Instant>,
}
impl CrashBudget {
/// 默认预算:60 秒窗口、5 次崩溃。
pub fn new() -> Self {
Self::with(CRASH_WINDOW, CRASH_LIMIT)
}
/// 指定窗口与上限构造预算(生产取默认值,测试注入短窗口)。
pub fn with(window: Duration, limit: usize) -> Self {
CrashBudget {
window,
limit,
times: Vec::new(),
}
}
/// 记录一次崩溃:清掉窗口外条目后追加,返回预算是否耗尽。
///
/// # 边界与前提
/// - 耗尽是幂等的:耗尽后继续记录仍返回耗尽。
pub fn record(&mut self, now: Instant) -> bool {
self.times.retain(|t| now.duration_since(*t) <= self.window);
self.times.push(now);
self.times.len() > self.limit
}
}
impl Default for CrashBudget {
fn default() -> Self {
Self::new()
}
}
/// watchdog 主循环:拉起 server、就绪探测、监控循环、崩溃重启、信号退出。
///
/// # 步骤
/// 1. 注册 ctrlc 处理器置位停止标志(SIGINT/SIGTERM,termination 特性)。
/// 须在初始 spawn 之前注册:sidecar 出现早于 socket 绑定,外部信号可能
/// 在就绪探测期间到达,处理器未就位时走默认终止行为。
/// 2. 初始 spawn(共享入口)加就绪探测(10 秒),失败打印退 1;成功打印
/// 监控开始标记行(外部崩溃模拟与信号在此之后才安全)。
/// 3. 监控循环(500 毫秒间隔):停止标志置位即转优雅关闭;收领子进程
/// (被杀的子进程是僵尸,PID 探活对僵尸误判活,子进程退出状态是死亡的
/// 权威信号);子进程即监控对象时按其存活判 Alive、已退出按 sidecar
/// 残留判 Crash;子进程在认领阶段被拒时监控对象是既有 server,按 decide
/// 三信号裁决;sidecar 连续两次(1 秒)判缺失即 server 已正常退出、退 0。
/// 4. 崩溃处理:打印崩溃提示(含进程号)、读日志文件最后 20 行打印、记预算
/// (耗尽则打印放弃提示退 1)、退避 1 秒、经同一 spawn 入口重启加就绪
/// 探测(失败打印退 1)、继续循环。
/// 5. 优雅关闭:尽力发关闭请求(忽略错误,server 可能已死),打印退 0。
///
/// # 边界与前提
/// - sidecar 缺失须连续两次轮询(1 秒)确认才判 Exited:新 server 的认领
/// 路径先删旧 sidecar 再重写,删除与重写之间是微秒级窗口,单次轮询落入
/// 窗口会误判「server 已正常退出」而静默退出、留下无人监控的 server;
/// 两次确认把该窗口压到实际不可达。
/// - 已有存活 server 时再跑 watchdog:spawn 的子进程在认领阶段被拒(同版本
/// 判 `AlreadyRunning`,退 1、不触碰文件),就绪探测看到第一个 server 即
/// 通过,watchdog 监控的是第一个 server(与双启动幂等同一语义,天然安全)。
/// - 新 server 启动即崩溃(绑定前消亡、sidecar 从未出现):就绪探测 10 秒
/// 超时后 watchdog 退 1(自限,不形成无限循环),消息带 `oxide server
/// cleanup` 人工检查提示。
///
/// # 副作用
/// - spawn 脱离的 server 子进程;崩溃后经同一入口重启。
/// - SIGINT/SIGTERM 时尽力向 server 发关闭请求。
pub fn run_watchdog(workers: Option<u32>) -> ExitCode {
// 先注册信号处理器:信号到达时置位停止标志(处理器体只做一次原子写)。
// 须在初始 spawn 之前注册:sidecar 出现早于 socket 绑定,外部信号可能
// 在就绪探测期间到达,处理器未就位时 SIGINT 走默认终止行为。
let stop = Arc::new(AtomicBool::new(false));
let flag = Arc::clone(&stop);
if let Err(err) = ctrlc::set_handler(move || {
flag.store(true, Ordering::SeqCst);
}) {
eprintln!("注册信号处理器失败:{err}");
return ExitCode::FAILURE;
}
// 初始 spawn 与就绪探测:失败即退 1,不进入监控循环。
let mut child = match spawn_detached_server(workers) {
Ok(child) => child,
Err(err) => {
eprintln!("spawn server 进程失败:{err}");
return ExitCode::FAILURE;
}
};
if let Err(msg) = wait_server_ready(READY_DEADLINE) {
eprintln!("{msg}(可用 `oxide server cleanup` 人工检查)");
return ExitCode::FAILURE;
}
// 初始就绪探测通过:外部信号(崩溃模拟、SIGINT)在此之后才安全,
// 该行同时是监控循环开始的标记。
println!("server 已就绪,watchdog 开始监控");
let mut budget = CrashBudget::new();
let sidecar_path = sidecar::well_known_sidecar_path();
let log_path = sidecar::well_known_log_path();
let mut missing_streak = 0usize;
// 子进程退出状态:收领后只记一次(try_wait 收领后不再重复出状态)。
let mut child_exited: Option<ExitStatus> = None;
loop {
// 停止标志置位:转优雅关闭(尽力停 server,退 0)。
if stop.load(Ordering::SeqCst) {
return graceful_stop();
}
std::thread::sleep(POLL_INTERVAL);
// 收领子进程:子进程是监控对象时,其退出状态是死亡的权威信号
// (被杀的子进程是僵尸,kill -0 对僵尸返回活,PID 探活会误判)。
if child_exited.is_none() {
if let Some(status) = child.try_wait().ok().flatten() {
child_exited = Some(status);
}
}
// 读 sidecar 采集信号:缺失要连续两次确认(认领路径的删旧写新窗口
// 是微秒级,单次轮询落入窗口会误判正常退出)。
let identity = sidecar::read_identity(&sidecar_path);
let state = match &identity {
None => {
missing_streak += 1;
if missing_streak >= 2 {
println!("server 已正常退出,watchdog 退出");
return ExitCode::SUCCESS;
}
continue;
}
Some(id) => {
missing_streak = 0;
if id.pid == child.id() {
// 子进程即监控对象:存活判 Alive,已退出按 sidecar 残留
// 区分崩溃与正常退出。
match child_exited {
Some(_) => WatchState::Crash,
None => WatchState::Alive,
}
} else {
// 子进程在认领阶段被拒(已有存活 server):监控对象是
// 既有 server,按三信号裁决。
let pid_alive = sidecar::pid_alive(id.pid);
let starttime_match = sidecar::process_starttime(id.pid) == Some(id.starttime);
decide(Some(id), pid_alive, starttime_match)
}
}
};
match state {
WatchState::Alive => {}
WatchState::Crash => {
// 崩溃处理:诊断打印、记预算、退避、经同一入口重启。
// 崩溃态下 sidecar 必可读(Crash 只从 Some 分支产生),进程号
// 取 sidecar 记录;读不到时以 0 占位(防御性兜底,不 panic)。
let pid = identity.as_ref().map(|id| id.pid).unwrap_or(0);
println!("server 已崩溃(进程号 {pid}),准备重启");
print_crash_log_tail(&log_path);
if budget.record(Instant::now()) {
eprintln!(
"崩溃预算耗尽({} 秒窗口内 {} 次崩溃),放弃自动重启,可用 `oxide server cleanup` 检查",
CRASH_WINDOW.as_secs(),
CRASH_LIMIT
);
return ExitCode::FAILURE;
}
std::thread::sleep(CRASH_BACKOFF);
match spawn_detached_server(workers) {
Ok(new_child) => {
child = new_child;
child_exited = None;
}
Err(err) => {
eprintln!("重启 server 失败:{err}");
return ExitCode::FAILURE;
}
}
if let Err(msg) = wait_server_ready(READY_DEADLINE) {
eprintln!("新 {msg}(可用 `oxide server cleanup` 人工检查)");
return ExitCode::FAILURE;
}
// 重启就绪探测通过:新 server 已可服务,外部操作(关闭请求)
// 在此之后才安全;该行与初始标记同文,是监控循环恢复的标记。
println!("server 已就绪,watchdog 开始监控");
}
WatchState::Exited => {
// 防御性兜底:sidecar 缺失分支已两次确认后返回,到达此处即
// 信号采集与裁决不一致。
println!("server 已正常退出,watchdog 退出");
return ExitCode::SUCCESS;
}
}
}
}
/// 打印日志文件最后 20 行(崩溃诊断);文件缺失或为空不视为错误。
///
/// # 边界与前提
/// - server 可能在写日志前就崩了,空日志与缺失文件都只跳过打印。
fn print_crash_log_tail(log_path: &Path) {
let lines = server_log::read_log(log_path, None, Some(CRASH_LOG_LINES)).unwrap_or_default();
if lines.is_empty() {
return;
}
println!("日志尾部(最后 {CRASH_LOG_LINES} 行):");
for line in lines {
println!("{line}");
}
}
/// 优雅关闭:尽力向 server 发关闭请求(忽略错误,server 可能已死),打印退 0。
fn graceful_stop() -> ExitCode {
let _ = send_control_request(&ServerRequest::Shutdown);
println!("server 已停止,watchdog 退出");
ExitCode::SUCCESS
}
#[cfg(test)]
mod tests {
use super::*;
/// 测试身份:字段为占位值(decide 只看 Option 与两个布尔信号)。
fn test_identity() -> ServerIdentity {
ServerIdentity {
pid: 1,
socket_path: "/tmp/oxide-test.sock".into(),
version: "0.0.0".into(),
started_at: 0,
starttime: 0,
}
}
/// decide 四行裁决表:sidecar 缺失判 Exited、PID 死判 Crash、PID 活刻度
/// 匹配判 Alive、PID 活刻度失配判 Crash。
#[test]
fn decide_four_row_table() {
let id = test_identity();
assert_eq!(decide(None, false, false), WatchState::Exited, "sidecar 缺失应判 Exited");
assert_eq!(decide(Some(&id), false, false), WatchState::Crash, "PID 死应判 Crash");
assert_eq!(decide(Some(&id), true, true), WatchState::Alive, "PID 活刻度匹配应判 Alive");
assert_eq!(decide(Some(&id), true, false), WatchState::Crash, "PID 活刻度失配应判 Crash");
}
/// 预算窗口语义:上限内未耗尽,窗口外条目被清后不再算入。
#[test]
fn budget_window_expiry() {
let mut budget = CrashBudget::with(Duration::from_millis(100), 2);
let t0 = Instant::now();
assert!(!budget.record(t0), "首次记录不应耗尽");
assert!(!budget.record(t0), "窗口内未超上限不应耗尽");
// 窗口过期后旧条目被清:新记录不再与旧条目叠加。
std::thread::sleep(Duration::from_millis(150));
let t1 = t0 + Duration::from_millis(150);
assert!(!budget.record(t1), "窗口外条目应被清,不应耗尽");
}
/// 第 6 次崩溃判耗尽(默认窗口 5 次上限)。
#[test]
fn budget_exhausted_on_sixth() {
let mut budget = CrashBudget::new();
let start = Instant::now();
for _ in 0..CRASH_LIMIT {
assert!(!budget.record(start), "上限内不应耗尽");
}
assert!(budget.record(start), "第 6 次应判耗尽");
}
/// 预算耗尽后的 record 仍返回耗尽(幂等)。
#[test]
fn budget_exhausted_is_idempotent() {
let mut budget = CrashBudget::with(Duration::from_secs(60), 2);
let start = Instant::now();
budget.record(start);
budget.record(start);
assert!(budget.record(start), "第 3 次应判耗尽");
assert!(budget.record(start), "耗尽后 record 应仍返回耗尽");
}
/// 记录时刻单调推进时窗口滑动:仅窗口内条目计入。
#[test]
fn budget_sliding_window_counts_only_in_window() {
let mut budget = CrashBudget::with(Duration::from_millis(100), 3);
let t0 = Instant::now();
budget.record(t0);
budget.record(t0);
std::thread::sleep(Duration::from_millis(150));
let t1 = t0 + Duration::from_millis(150);
// 窗口外旧条目被清:窗口内 3 条恰达上限不耗尽,第 4 条耗尽。
for _ in 0..3 {
assert!(!budget.record(t1), "窗口内未超上限不应耗尽");
}
assert!(budget.record(t1), "窗口内第 4 条应判耗尽");
}
}
+183 -6
View File
@@ -1,16 +1,18 @@
//! server CLI 集成测试:真实二进制的分派面(help 列子命令、无 server 时 //! server CLI 集成测试:真实二进制的分派面(help 列子命令、无 server 时
//! 控制臂与 forge 退 1、cleanup 幂等、start 与 restart 端到端、forge 端到端)。 //! 控制臂与 forge 退 1、cleanup 幂等、start 与 restart 端到端、forge 端到端、
//! watchdog 崩溃自动重启与 SIGINT 优雅停 server)。
//! //!
//! start / restart 端到端与 cleanup 走 well-known 全局路径(每用户单例): //! start / restart / watchdog 端到端与 cleanup 走 well-known 全局路径
//! 测试先探活,socket 存活即 panic 不抢占存活 server;三枚触碰全局路径的 //! (每用户单例):测试先探活,socket 存活即 panic 不抢占存活 server;
//! 测试经同一把锁串行,避免相互干扰。 //! 触碰全局路径的测试经同一把锁串行,避免相互干扰。
use std::io::{BufReader, Write}; use std::io::{BufRead, BufReader, Write};
use std::os::unix::net::UnixStream; use std::os::unix::net::UnixStream;
use std::process::{Command, Output}; use std::process::{Command, Output};
use std::sync::Mutex; use std::sync::Mutex;
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use oxide_cli::server::client::send_control_request;
use oxide_cli::server::protocol::{self, FrameReader, ServerRequest, ServerResponse}; use oxide_cli::server::protocol::{self, FrameReader, ServerRequest, ServerResponse};
use oxide_cli::server::sidecar; use oxide_cli::server::sidecar;
@@ -25,6 +27,31 @@ fn oxide(args: &[&str]) -> Output {
.expect("failed to run oxide") .expect("failed to run oxide")
} }
/// 向进程发信号(与 sidecar liveness 探活同一 `kill` 二进制路径)。
fn kill_process(pid: u32, signal: &str) {
Command::new("kill")
.args([signal, &pid.to_string()])
.status()
.expect("发送信号应成功");
}
/// 轮询健康请求直至得 healthy(100 毫秒间隔、给定截止)。
///
/// sidecar 出现早于 socket 绑定与池预热,不能直接作为就绪信号;watchdog
/// 的初始就绪探测与测试的崩溃模拟都以健康请求通过为界。
fn wait_for_healthy(deadline: Duration) {
let end = Instant::now() + deadline;
loop {
let ready =
matches!(send_control_request(&ServerRequest::Health), Ok(ServerResponse::Health { healthy: true }));
if ready {
return;
}
assert!(Instant::now() < end, "server 应在有界时间内就绪");
std::thread::sleep(Duration::from_millis(100));
}
}
/// 发送请求帧并读回一帧响应。 /// 发送请求帧并读回一帧响应。
fn send_and_recv(stream: &mut UnixStream, request: &ServerRequest) -> ServerResponse { fn send_and_recv(stream: &mut UnixStream, request: &ServerRequest) -> ServerResponse {
stream stream
@@ -35,7 +62,7 @@ fn send_and_recv(stream: &mut UnixStream, request: &ServerRequest) -> ServerResp
protocol::parse_response(&frame).expect("响应帧解析应成功") protocol::parse_response(&frame).expect("响应帧解析应成功")
} }
/// help 列全部十个子命令。 /// help 列全部十一个子命令。
#[test] #[test]
fn server_help_lists_subcommands() { fn server_help_lists_subcommands() {
let output = oxide(&["server", "--help"]); let output = oxide(&["server", "--help"]);
@@ -43,6 +70,7 @@ fn server_help_lists_subcommands() {
let stdout = String::from_utf8_lossy(&output.stdout); let stdout = String::from_utf8_lossy(&output.stdout);
for name in [ for name in [
"start", "stop", "status", "health", "info", "version", "cleanup", "restart", "log", "forge", "start", "stop", "status", "health", "info", "version", "cleanup", "restart", "log", "forge",
"watchdog",
] { ] {
assert!(stdout.contains(name), "help 应列出 {name}:{stdout}"); assert!(stdout.contains(name), "help 应列出 {name}:{stdout}");
} }
@@ -432,3 +460,152 @@ fn server_forge_e2e() {
assert!(!socket.exists(), "socket 文件应被删除"); assert!(!socket.exists(), "socket 文件应被删除");
assert!(!sidecar_path.exists(), "sidecar 文件应被删除"); assert!(!sidecar_path.exists(), "sidecar 文件应被删除");
} }
/// watchdog 端到端:拉起 watchdog 前台进程(其拉起 server),`kill -9` 模拟
/// 崩溃,断言 watchdog 自动重启(新 sidecar 进程号不同、新 server 健康),
/// 关闭后 watchdog 退 0 且 stdout 含崩溃提示与日志尾部。
#[test]
fn server_watchdog_e2e() {
let _guard = WELL_KNOWN_LOCK.lock().expect("全局路径锁不应中毒");
let socket = sidecar::well_known_socket_path();
let sidecar_path = sidecar::well_known_sidecar_path();
// 探活先行:存活 server 占用全局路径时 panic,不抢占。
assert!(!sidecar::is_server_alive(&socket), "存活 server 占用 well-known 路径,测试不抢占");
// 拉起 watchdog 前台进程(stdout 管道捕获崩溃诊断输出)。
let mut child = Command::new(env!("CARGO_BIN_EXE_oxide"))
.args(["server", "watchdog", "--workers", "2"])
.stdout(std::process::Stdio::piped())
.spawn()
.expect("oxide server watchdog 应可启动");
// 后台线程消费 stdout 收集全部行,见到「开始监控」标记行即通知:
// 该标记行蕴含 watchdog 已越过初始就绪探测,崩溃模拟在此之后进行
// 才不与初始探测窗口竞争(预热期间杀 server 会让探测超时退 1、不重启)。
let stdout_pipe = child.stdout.take().expect("stdout 管道应可取");
let (monitoring_tx, monitoring_rx) = std::sync::mpsc::channel::<()>();
let out_handle = std::thread::spawn(move || {
let mut reader = BufReader::new(stdout_pipe);
let mut lines = Vec::new();
let mut line = String::new();
loop {
line.clear();
if reader.read_line(&mut line).unwrap_or(0) == 0 {
break;
}
lines.push(line.clone());
if line.contains("watchdog 开始监控") {
let _ = monitoring_tx.send(());
}
}
lines
});
// 轮询 sidecar 出现(10 秒截止)判 server 已启动。
let deadline = Instant::now() + Duration::from_secs(10);
while sidecar::read_identity(&sidecar_path).is_none() {
assert!(Instant::now() < deadline, "sidecar 未出现");
std::thread::sleep(Duration::from_millis(50));
}
let first_id = sidecar::read_identity(&sidecar_path).expect("读 sidecar 应成功");
// 等 watchdog 越过初始就绪探测进入监控循环(30 秒截止)。
monitoring_rx
.recv_timeout(Duration::from_secs(30))
.expect("watchdog 应在有界时间内进入监控循环");
// 模拟崩溃:kill -9(文件残留、无退出序列,与 panic=abort 消亡等价)。
kill_process(first_id.pid, "-9");
// 轮询新 sidecar 出现且进程号不同(30 秒截止、50 毫秒间隔)判 watchdog
// 已重启。
let deadline = Instant::now() + Duration::from_secs(30);
let second_id = loop {
if let Some(id) = sidecar::read_identity(&sidecar_path) {
if id.pid != first_id.pid {
break id;
}
}
assert!(Instant::now() < deadline, "watchdog 应在有界时间内重启 server");
std::thread::sleep(Duration::from_millis(50));
};
// 等 watchdog 越过重启就绪探测(30 秒截止):关闭请求发给探测完成前的
// 新 server 会让它优雅退出、删 sidecar,与 watchdog 的重启探测竞争
// (探测超时退 1、不重启)。
monitoring_rx
.recv_timeout(Duration::from_secs(30))
.expect("watchdog 应在有界时间内完成重启探测");
// 连接新 server:健康请求得 healthy。
let mut stream = UnixStream::connect(&socket).expect("连接新 server 应成功");
match send_and_recv(&mut stream, &ServerRequest::Health) {
ServerResponse::Health { healthy } => assert!(healthy, "新 server 应健康"),
other => panic!("应得 Health 帧,实得 {other:?}"),
}
// 发关闭请求得确认帧,轮询 sidecar 消失(watchdog 判 Exited 退 0)。
let response = send_and_recv(&mut stream, &ServerRequest::Shutdown);
assert!(matches!(response, ServerResponse::Shutdown), "应得关闭确认帧:{response:?}");
drop(stream);
let deadline = Instant::now() + Duration::from_secs(30);
while sidecar_path.exists() {
assert!(Instant::now() < deadline, "server 应在有界时间内退出");
std::thread::sleep(Duration::from_millis(50));
}
// 等 watchdog 子进程退出断言退出码 0,收集 stdout 断言含崩溃提示与日志尾部。
let status = child.wait().expect("等待 watchdog 应成功");
assert_eq!(status.code(), Some(0), "watchdog 应以退出码 0 退出:{status:?}");
let lines = out_handle.join().expect("stdout 读取线程应可汇合");
let stdout = lines.join("");
assert!(stdout.contains("server 已崩溃"), "watchdog 应打印崩溃提示:{stdout}");
assert!(stdout.contains("日志尾部"), "watchdog 应打印日志尾部:{stdout}");
assert_ne!(second_id.pid, first_id.pid, "重启后的 server 进程号应不同");
}
/// watchdog SIGINT:拉起 watchdog 前台进程后向其发 SIGINT,断言 watchdog 退 0、
/// sidecar 消失、server 进程已退出(停 watchdog 即优雅停 server)。
#[test]
fn server_watchdog_sigint_stops_server() {
let _guard = WELL_KNOWN_LOCK.lock().expect("全局路径锁不应中毒");
let socket = sidecar::well_known_socket_path();
let sidecar_path = sidecar::well_known_sidecar_path();
// 探活先行:存活 server 占用全局路径时 panic,不抢占。
assert!(!sidecar::is_server_alive(&socket), "存活 server 占用 well-known 路径,测试不抢占");
// 拉起 watchdog 前台进程。
let mut child = Command::new(env!("CARGO_BIN_EXE_oxide"))
.args(["server", "watchdog", "--workers", "2"])
.spawn()
.expect("oxide server watchdog 应可启动");
// 轮询 sidecar 出现(10 秒截止)判 server 已启动。
let deadline = Instant::now() + Duration::from_secs(10);
while sidecar::read_identity(&sidecar_path).is_none() {
assert!(Instant::now() < deadline, "sidecar 未出现");
std::thread::sleep(Duration::from_millis(50));
}
let id = sidecar::read_identity(&sidecar_path).expect("读 sidecar 应成功");
// 等健康请求通过判 server 就绪:watchdog 已越过初始就绪探测进入监控
// 循环,SIGINT 的优雅关闭请求发给已就绪的 server 即被处理。
wait_for_healthy(Duration::from_secs(60));
// 向 watchdog 发 SIGINT。
kill_process(child.id(), "-INT");
// 等 watchdog 退出断言退出码 0。
let status = child.wait().expect("等待 watchdog 应成功");
assert_eq!(status.code(), Some(0), "watchdog 应以退出码 0 退出:{status:?}");
// server 被优雅停止:sidecar 消失、进程已退出。
let deadline = Instant::now() + Duration::from_secs(30);
while sidecar_path.exists() {
assert!(Instant::now() < deadline, "server 应在有界时间内退出");
std::thread::sleep(Duration::from_millis(50));
}
assert!(!sidecar::pid_alive(id.pid), "server 进程应已退出");
}