diff --git a/Cargo.lock b/Cargo.lock index 75d9358e..4d2eef23 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.8.41" +version = "0.8.42" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.8.41" +version = "0.8.42" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.8.41" +version = "0.8.42" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.8.41" +version = "0.8.42" dependencies = [ "ansi_term", "bumpalo", @@ -1225,7 +1225,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.8.41" +version = "0.8.42" dependencies = [ "lru", "oxide_bytecode", @@ -1234,7 +1234,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.8.41" +version = "0.8.42" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1253,7 +1253,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.8.41" +version = "0.8.42" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1267,7 +1267,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.8.41" +version = "0.8.42" dependencies = [ "num-bigint", "oxc_allocator", @@ -1282,7 +1282,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.8.41" +version = "0.8.42" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1291,7 +1291,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.8.41" +version = "0.8.42" dependencies = [ "bumpalo", "dashmap", @@ -1306,7 +1306,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.8.41" +version = "0.8.42" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1318,7 +1318,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.8.41" +version = "0.8.42" dependencies = [ "tracing", "tracing-appender", @@ -1327,7 +1327,7 @@ dependencies = [ [[package]] name = "oxide_parser" -version = "0.8.41" +version = "0.8.42" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1342,7 +1342,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.8.41" +version = "0.8.42" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1358,7 +1358,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.8.41" +version = "0.8.42" dependencies = [ "num-bigint", "num-traits", @@ -1370,7 +1370,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.8.41" +version = "0.8.42" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1387,7 +1387,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.8.41" +version = "0.8.42" dependencies = [ "bumpalo", "hashbrown 0.14.5", @@ -1399,7 +1399,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.8.41" +version = "0.8.42" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 97f13c61..05918887 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ members = [ ] [workspace.package] -version = "0.8.41" +version = "0.8.42" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_cli/src/main.rs b/crates/oxide_cli/src/main.rs index c4994364..9ff61b8d 100644 --- a/crates/oxide_cli/src/main.rs +++ b/crates/oxide_cli/src/main.rs @@ -2,7 +2,7 @@ #![allow(dead_code)] use std::fs; -use std::process::{Command, ExitCode, Stdio}; +use std::process::ExitCode; use std::sync::Arc; 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::server::{run_server, run_server_rm, RmServerConfig, ServerConfig}; 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_error; use oxide_kernel::kernel::{KernelConfig, KernelCore}; @@ -169,6 +171,12 @@ enum ServerCommands { #[arg(long)] lookup: Option, }, + /// 监控 server 存活并在崩溃后自动重启（前台进程，60 秒窗口 5 次崩溃即停）。 + Watchdog { + /// 常驻 worker 线程数；缺省取宿主核数。 + #[arg(long)] + workers: Option, + }, } /// forge 查询目标（CLI 侧，映射到协议侧 `ForgeTarget`）。 @@ -281,10 +289,8 @@ fn main() -> ExitCode { } } -/// server 子命令分派：start（两形态）与 cleanup 为最小实现， -/// version/status/health/info/stop/restart 六臂为真实实现， -/// log 为真实实现（读 well-known 日志文件，支持 --follow/--level/--lines）， -/// forge 为真实实现（查询四张共享 forge 状态，加三行动旗标）。 +/// server 子命令分派：全部十一臂为真实实现（start 两形态、cleanup、 +/// 五个控制臂、restart、log、forge、watchdog）。 /// /// # 步骤 /// 1. Start：rm 为真调独立模式入口（进程唯一路径）；foreground 为真当前进程 @@ -301,6 +307,8 @@ fn main() -> ExitCode { /// `--follow` 时进入阻塞跟踪循环（详见 log_command）。 /// 6. Forge：发 forge 查询请求（目标加三行动旗标），成功打印渲染文本退 0， /// 失败打印错误消息退 1。 +/// 7. Watchdog：前台监控进程，拉起 server、三态裁决、崩溃后自动重启 +/// （详见 run_watchdog）。 /// /// # 边界与前提 /// - `--idle-timeout` 不带 --rm 时静默忽略（语义门控归后续任务）。 @@ -311,6 +319,7 @@ fn main() -> ExitCode { /// 收领继续运行）。 /// - cleanup 可能向 sidecar 记录的进程发 SIGTERM 并删除残留文件。 /// - 控制臂向 server 建立并关闭一条 Unix socket 连接。 +/// - watchdog spawn 脱离的 server 子进程，崩溃后经同一入口重启。 fn server_command(command: ServerCommands) -> ExitCode { match command { ServerCommands::Start { rm, idle_timeout, workers, foreground } => { @@ -411,6 +420,7 @@ fn server_command(command: ServerCommands) -> ExitCode { _ => None, }) } + ServerCommands::Watchdog { workers } => watchdog::run_watchdog(workers), } } @@ -484,13 +494,13 @@ fn control_arm(request: &ServerRequest, render: fn(&ServerResponse) -> Option) -> ExitCode { // 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}"))); return ExitCode::FAILURE; } // 就绪探测：单循环全覆盖「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` 人工检查）"))); return ExitCode::FAILURE; } @@ -499,60 +509,6 @@ fn start_daemon(workers: Option) -> ExitCode { 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) -> std::io::Result { - 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 进程。 /// /// # 步骤 @@ -591,13 +547,13 @@ fn restart_command() -> ExitCode { } // 拉起新 server：脱离进程，标准流全空，不等待。 - if let Err(err) = spawn_detached_server(None) { + if let Err(err) = spawn::spawn_detached_server(None) { eprintln!("{}", Red.paint(format!("拉起新 server 失败：{err}"))); return ExitCode::FAILURE; } // 就绪探测：健康请求通过即新 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}"))); return ExitCode::FAILURE; } diff --git a/crates/oxide_cli/src/server/mod.rs b/crates/oxide_cli/src/server/mod.rs index 67de9432..4950891e 100644 --- a/crates/oxide_cli/src/server/mod.rs +++ b/crates/oxide_cli/src/server/mod.rs @@ -2,14 +2,16 @@ //! //! 持久 server 是常驻进程，复用预热池、近零 spawn 成本； //! 通信走 Unix socket + NDJSON（换行分隔 JSON）。 -//! 模块分八层：协议数据面（帧格式与 serde 结构体）、身份注册（sidecar 文件 +//! 模块分十层：协议数据面（帧格式与 serde 结构体）、身份注册（sidecar 文件 //! 与 liveness 检查原语）、控制客户端（读 sidecar、连接 server、发控制请求 //! 帧、读回响应帧、三分支错误）、执行请求路径（parse → compile → spawn → //! run → format → drop 六段纯函数）、worker 池（固定 N 常驻 worker、各持 //! 自有 VM 池、mpsc 轮询路由）、server 进程主体（accept 循环、控制请求直接 //! 处理、执行请求路由、优雅退出）、日志读取（行级别解析、阈值加最后 N 行 //! 过滤、阻塞式跟踪）、forge 状态查询（读四张共享 forge 条目数与容量、执行 -//! gc / clear-cache / lookup 三旗标）。 +//! gc / clear-cache / lookup 三旗标）、spawn 与就绪探测（start、restart、 +//! watchdog 三条路径共用的唯一共享入口）、watchdog（前台监控进程、三态裁决 +//! 加崩溃预算、崩溃后自动重启）。 /// 人工兜底清理命令：扫 sidecar、探活、杀 PID、删孤儿文件。 pub mod cleanup; @@ -34,5 +36,9 @@ pub mod protocol; pub mod server; /// 身份注册：sidecar 文件、liveness 检查与陈旧清理。 pub mod sidecar; +/// spawn 与就绪探测：start、restart、watchdog 三条路径共用的唯一共享入口。 +pub mod spawn; /// worker 池：固定 N 常驻 worker、各持自有 VM 池、mpsc 轮询路由。 pub mod workers; +/// watchdog：前台监控进程，三态裁决加崩溃预算，崩溃后自动重启。 +pub mod watchdog; diff --git a/crates/oxide_cli/src/server/spawn.rs b/crates/oxide_cli/src/server/spawn.rs new file mode 100644 index 00000000..4b094486 --- /dev/null +++ b/crates/oxide_cli/src/server/spawn.rs @@ -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) -> std::io::Result { + 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)); + } +} diff --git a/crates/oxide_cli/src/server/watchdog.rs b/crates/oxide_cli/src/server/watchdog.rs new file mode 100644 index 00000000..8a09f1ea --- /dev/null +++ b/crates/oxide_cli/src/server/watchdog.rs @@ -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, +} + +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) -> 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 = 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 条应判耗尽"); + } +} diff --git a/crates/oxide_cli/tests/server_cli.rs b/crates/oxide_cli/tests/server_cli.rs index 4fcdc592..39c1ac3c 100644 --- a/crates/oxide_cli/tests/server_cli.rs +++ b/crates/oxide_cli/tests/server_cli.rs @@ -1,16 +1,18 @@ //! server CLI 集成测试：真实二进制的分派面（help 列子命令、无 server 时 -//! 控制臂与 forge 退 1、cleanup 幂等、start 与 restart 端到端、forge 端到端）。 +//! 控制臂与 forge 退 1、cleanup 幂等、start 与 restart 端到端、forge 端到端、 +//! watchdog 崩溃自动重启与 SIGINT 优雅停 server）。 //! -//! start / restart 端到端与 cleanup 走 well-known 全局路径（每用户单例）： -//! 测试先探活，socket 存活即 panic 不抢占存活 server；三枚触碰全局路径的 -//! 测试经同一把锁串行，避免相互干扰。 +//! start / restart / watchdog 端到端与 cleanup 走 well-known 全局路径 +//! （每用户单例）：测试先探活，socket 存活即 panic 不抢占存活 server； +//! 触碰全局路径的测试经同一把锁串行，避免相互干扰。 -use std::io::{BufReader, Write}; +use std::io::{BufRead, BufReader, Write}; use std::os::unix::net::UnixStream; use std::process::{Command, Output}; use std::sync::Mutex; 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::sidecar; @@ -25,6 +27,31 @@ fn oxide(args: &[&str]) -> Output { .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 { stream @@ -35,7 +62,7 @@ fn send_and_recv(stream: &mut UnixStream, request: &ServerRequest) -> ServerResp protocol::parse_response(&frame).expect("响应帧解析应成功") } -/// help 列全部十个子命令。 +/// help 列全部十一个子命令。 #[test] fn server_help_lists_subcommands() { let output = oxide(&["server", "--help"]); @@ -43,6 +70,7 @@ fn server_help_lists_subcommands() { let stdout = String::from_utf8_lossy(&output.stdout); for name in [ "start", "stop", "status", "health", "info", "version", "cleanup", "restart", "log", "forge", + "watchdog", ] { assert!(stdout.contains(name), "help 应列出 {name}：{stdout}"); } @@ -432,3 +460,152 @@ fn server_forge_e2e() { assert!(!socket.exists(), "socket 文件应被删除"); 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 进程应已退出"); +}