mirror of
https://github.com/pjh456/OxideJS.git
synced 2026-10-23 18:21:13 +08:00
feat: server 身份注册，sidecar 原子创建与 liveness 检查原语
This commit is contained in:
@@ -1,9 +1,12 @@
|
||||
//! 持久 server 子系统：当前仅含协议数据面。
|
||||
//! 持久 server 子系统：当前含协议数据面与身份注册。
|
||||
//!
|
||||
//! 持久 server 是常驻进程，复用预热池、近零 spawn 成本；
|
||||
//! 通信走 Unix socket + NDJSON（换行分隔 JSON）。
|
||||
//! server 骨架、accept 循环与执行路径在后续任务，本模块先立协议数据面
|
||||
//! （帧格式与 serde 结构体），供 server 与客户端共用。
|
||||
//! （帧格式与 serde 结构体）与身份注册（sidecar 文件与 liveness 检查原语），
|
||||
//! 供 server 与客户端共用。
|
||||
|
||||
/// 协议数据面：NDJSON 帧格式与 serde 结构体。
|
||||
pub mod protocol;
|
||||
/// 身份注册：sidecar 文件、liveness 检查与陈旧清理。
|
||||
pub mod sidecar;
|
||||
|
||||
@@ -0,0 +1,416 @@
|
||||
//! server 身份注册：sidecar 文件（单行 JSON）与 liveness 检查原语。
|
||||
//!
|
||||
//! 关键约定：
|
||||
//! - sidecar 是启动时写入一次的静态身份快照，不记池状态（动态状态由控制协议
|
||||
//! 实时查询，写进 sidecar 只会过期）。
|
||||
//! - 单例性由内核 O_EXCL 创建标志保证（文件已存在时创建失败），不需要锁。
|
||||
//! - liveness 以 socket 连接探活为唯一裁决（内核只把连接完成给存活监听者），
|
||||
//! PID 探活为次信号，PID 复用以启动时刻刻度交叉核对排除。
|
||||
//! - --rm 模式走独立分支：生成唯一 socket 路径、不写 sidecar、不碰全局
|
||||
//! well-known 路径；该不变量由运行路径执行。
|
||||
|
||||
use std::fs::{self, OpenOptions};
|
||||
use std::io::{self, Write};
|
||||
use std::os::unix::net::UnixStream;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::Command;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// server 身份快照：单行 JSON sidecar 文件，启动时写入一次。
|
||||
///
|
||||
/// 字段：
|
||||
/// - `pid`：server 进程标识符，清理命令杀进程用。
|
||||
/// - `socket_path`：well-known socket 的绝对路径，控制客户端连接用。
|
||||
/// - `version`：构建期版本，版本交接时比对用。
|
||||
/// - `started_at`：Unix 秒，人读展示用。
|
||||
/// - `starttime`：`/proc/<pid>/stat` 的启动时刻刻度（自开机单调递增），
|
||||
/// PID 复用判定用。
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ServerIdentity {
|
||||
pub pid: u32,
|
||||
pub socket_path: String,
|
||||
pub version: String,
|
||||
pub started_at: u64,
|
||||
pub starttime: u64,
|
||||
}
|
||||
|
||||
impl ServerIdentity {
|
||||
/// 构造当前进程的身份（pid、版本、启动时间、启动时刻刻度）。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - 非 Linux 宿主（`/proc` 缺位）时 `starttime` 填 0，交叉核对退化，
|
||||
/// liveness 由 socket 探活裁决。
|
||||
pub fn new(socket_path: impl AsRef<Path>) -> Self {
|
||||
let pid = std::process::id();
|
||||
let started_at = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
|
||||
ServerIdentity {
|
||||
pid,
|
||||
socket_path: socket_path.as_ref().to_string_lossy().into_owned(),
|
||||
version: env!("CARGO_PKG_VERSION").to_string(),
|
||||
started_at,
|
||||
starttime: process_starttime(pid).unwrap_or(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 读取当前进程 uid（`/proc/self/status` 的 `Uid:` 行首值）。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - 非 Linux 宿主或解析失败时返回 0，路径退化为固定名，由 liveness 检查兜底。
|
||||
pub fn current_uid() -> u32 {
|
||||
let status = match fs::read_to_string("/proc/self/status") {
|
||||
Ok(s) => s,
|
||||
Err(_) => return 0,
|
||||
};
|
||||
for line in status.lines() {
|
||||
if let Some(rest) = line.strip_prefix("Uid:") {
|
||||
if let Some(first) = rest.split_whitespace().next() {
|
||||
return first.parse().unwrap_or(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
0
|
||||
}
|
||||
|
||||
/// well-known sidecar 路径：`$TMPDIR/oxide-<uid>.json`。
|
||||
pub fn well_known_sidecar_path() -> PathBuf {
|
||||
std::env::temp_dir().join(format!("oxide-{}.json", current_uid()))
|
||||
}
|
||||
|
||||
/// well-known 持久 server socket 路径：`$TMPDIR/oxide-<uid>.sock`。
|
||||
pub fn well_known_socket_path() -> PathBuf {
|
||||
std::env::temp_dir().join(format!("oxide-{}.sock", current_uid()))
|
||||
}
|
||||
|
||||
/// O_EXCL 原子创建 sidecar（文件已存在时创建失败）。
|
||||
///
|
||||
/// # 步骤
|
||||
/// 1. 身份序列化为单行紧凑 JSON。
|
||||
/// 2. `create_new(true)` 打开并写入全部内容。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - 文件已存在时返回 `AlreadyExists`，调用方据此转入读回裁决。
|
||||
///
|
||||
/// # 注意事项
|
||||
/// - 写入不同步磁盘：崩溃残留的半写文件按损坏处理，由 liveness 检查兜底。
|
||||
pub fn write_exclusive(identity: &ServerIdentity, path: &Path) -> io::Result<()> {
|
||||
let json = serde_json::to_string(identity).expect("身份序列化不会失败");
|
||||
let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
|
||||
file.write_all(json.as_bytes())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 读回并解析 sidecar。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - 文件缺失或解析失败（损坏、半写）返回 `None`，调用方按陈旧残留处理。
|
||||
pub fn read_identity(path: &Path) -> Option<ServerIdentity> {
|
||||
let text = fs::read_to_string(path).ok()?;
|
||||
serde_json::from_str(&text).ok()
|
||||
}
|
||||
|
||||
/// socket 探活：连接成功即存在存活 server。
|
||||
///
|
||||
/// 内核只把连接完成给存活监听者，该信号不依赖 PID、不受 PID 复用影响，
|
||||
/// 是注册判定的唯一裁决。
|
||||
pub fn is_server_alive(socket_path: &Path) -> bool {
|
||||
UnixStream::connect(socket_path).is_ok()
|
||||
}
|
||||
|
||||
/// PID 探活：`kill -0` 退出码 0 即进程存在。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - `kill` 二进制缺失或进程不存在时返回 `false`。
|
||||
pub fn pid_alive(pid: u32) -> bool {
|
||||
Command::new("kill")
|
||||
.args(["-0", &pid.to_string()])
|
||||
.status()
|
||||
.map(|s| s.success())
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
/// 读取进程启动时刻刻度（`/proc/<pid>/stat` 第 22 字段，自开机单调递增）。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - 进程不存在或 `/proc` 缺位（非 Linux）时返回 `None`。
|
||||
pub fn process_starttime(pid: u32) -> Option<u64> {
|
||||
let text = fs::read_to_string(format!("/proc/{pid}/stat")).ok()?;
|
||||
// comm 字段带括号且可能含空格，从最后一个 `)` 起切分，避免 comm 内的
|
||||
// 空格干扰字段计数。
|
||||
let rest = text.rsplit(')').next()?;
|
||||
// 剩余部分从 state（全文件第 3 字段）起，第 20 个字段即全文件第 22 字段。
|
||||
rest.split_whitespace().skip(19).next()?.parse().ok()
|
||||
}
|
||||
|
||||
/// 认领结果：注册入口的三态。
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum ClaimResult {
|
||||
/// 新注册成功（sidecar 已写入）。
|
||||
Registered,
|
||||
/// 已有存活 server，拒绝注册。
|
||||
AlreadyRunning,
|
||||
/// 存活证据不足以裁决（进程存活但 socket 不在），保守拒绝并提示跑清理命令。
|
||||
RefusedAmbiguous,
|
||||
}
|
||||
|
||||
/// 注册入口：认领 well-known sidecar，必要时清理陈旧残留。
|
||||
///
|
||||
/// # 步骤
|
||||
/// 1. 尝试 O_EXCL 创建；sidecar 不存在即成功。
|
||||
/// 2. sidecar 已存在：探活 socket，连接成功即有存活 server，拒绝。
|
||||
/// 3. socket 死：按 sidecar 可读性分支。
|
||||
/// - 可解析且 PID 死：陈旧残留，删 sidecar 与 socket 文件后重建。
|
||||
/// - 可解析且 PID 活：交叉核对启动时刻刻度，匹配则保守拒绝（server
|
||||
/// 可能正在启动或 socket 被外部删除），失配则 PID 已复用，删除重建。
|
||||
/// - 不可解析（损坏）：只删 sidecar（socket 路径未知不猜）后重建。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - 启动顺序契约是「先写 sidecar 再 bind socket」，存活 server 必有 sidecar。
|
||||
///
|
||||
/// # 副作用
|
||||
/// - 可能创建、删除、重写 sidecar 与 socket 文件。
|
||||
///
|
||||
/// # 注意事项
|
||||
/// - 意外 io 错误（权限等）保守判 `RefusedAmbiguous`；方向安全（拒绝注册
|
||||
/// 不会双注册）。
|
||||
pub fn claim_sidecar(sidecar_path: &Path, socket_path: &Path) -> ClaimResult {
|
||||
// 先尝试 O_EXCL 创建：sidecar 不存在即成功。
|
||||
match write_exclusive(&ServerIdentity::new(socket_path), sidecar_path) {
|
||||
Ok(()) => return ClaimResult::Registered,
|
||||
Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}
|
||||
Err(_) => return ClaimResult::RefusedAmbiguous,
|
||||
}
|
||||
|
||||
// socket 探活是存活的唯一事实源：连接成功即有存活 server。
|
||||
if is_server_alive(socket_path) {
|
||||
return ClaimResult::AlreadyRunning;
|
||||
}
|
||||
|
||||
// socket 死：按 sidecar 可读性分支。
|
||||
match read_identity(sidecar_path) {
|
||||
Some(id) => claim_dead_socket(&id, sidecar_path, socket_path),
|
||||
None => {
|
||||
// 损坏文件：只删 sidecar（socket 路径未知不猜）后重建。
|
||||
let _ = fs::remove_file(sidecar_path);
|
||||
re_register(sidecar_path, socket_path)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// socket 死且 sidecar 可解析：按 PID 与启动时刻刻度裁决陈旧性。
|
||||
fn claim_dead_socket(id: &ServerIdentity, sidecar_path: &Path, socket_path: &Path) -> ClaimResult {
|
||||
// PID 死：陈旧残留，删 sidecar 与 socket 文件后重建。
|
||||
if !pid_alive(id.pid) {
|
||||
remove_stale(sidecar_path, socket_path);
|
||||
return re_register(sidecar_path, socket_path);
|
||||
}
|
||||
|
||||
// PID 活：交叉核对启动时刻刻度裁决 PID 复用。
|
||||
if process_starttime(id.pid) == Some(id.starttime) {
|
||||
// 进程活且刻度匹配：server 可能正在启动或 socket 被外部删除，保守拒绝。
|
||||
return ClaimResult::RefusedAmbiguous;
|
||||
}
|
||||
|
||||
// 刻度失配：PID 已被复用，sidecar 是陈旧残留。
|
||||
remove_stale(sidecar_path, socket_path);
|
||||
re_register(sidecar_path, socket_path)
|
||||
}
|
||||
|
||||
/// 删除 sidecar 与 socket 文件（socket 探活已失败，文件无监听者，不会误删
|
||||
/// 存活 server 的 socket）。
|
||||
fn remove_stale(sidecar_path: &Path, socket_path: &Path) {
|
||||
let _ = fs::remove_file(sidecar_path);
|
||||
let _ = fs::remove_file(socket_path);
|
||||
}
|
||||
|
||||
/// 清理后重建 sidecar，重建失败保守拒绝。
|
||||
fn re_register(sidecar_path: &Path, socket_path: &Path) -> ClaimResult {
|
||||
match write_exclusive(&ServerIdentity::new(socket_path), sidecar_path) {
|
||||
Ok(()) => ClaimResult::Registered,
|
||||
Err(_) => ClaimResult::RefusedAmbiguous,
|
||||
}
|
||||
}
|
||||
|
||||
/// 生成 --rm 模式唯一 socket 路径：`$TMPDIR/oxide-rm-<pid>-<纳秒时间戳>.sock`。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - PID 加纳秒时间戳在单进程内唯一；--rm 是独立单进程，不需要 O_EXCL 竞争。
|
||||
///
|
||||
/// # 注意事项
|
||||
/// - 不变量：--rm 模式从不调用 `claim_sidecar`、不写 sidecar、不碰全局
|
||||
/// well-known 路径。
|
||||
pub fn rm_socket_path() -> PathBuf {
|
||||
let ns = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos();
|
||||
std::env::temp_dir().join(format!("oxide-rm-{}-{ns}.sock", std::process::id()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::os::unix::net::UnixListener;
|
||||
use std::time::Duration;
|
||||
|
||||
/// temp_dir 下的唯一临时目录，退出时自动删除。
|
||||
struct TestDir(PathBuf);
|
||||
|
||||
impl TestDir {
|
||||
fn new() -> Self {
|
||||
let ns = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
|
||||
let dir = std::env::temp_dir().join(format!("oxide_sidecar_test_{}_{}", std::process::id(), ns));
|
||||
fs::create_dir_all(&dir).expect("测试目录创建失败");
|
||||
TestDir(dir)
|
||||
}
|
||||
|
||||
fn path(&self, name: &str) -> PathBuf {
|
||||
self.0.join(name)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TestDir {
|
||||
fn drop(&mut self) {
|
||||
let _ = fs::remove_dir_all(&self.0);
|
||||
}
|
||||
}
|
||||
|
||||
/// 保证不存在的 pid：pid_max 之上永远不会被分配。
|
||||
fn dead_pid() -> u32 {
|
||||
let pid_max = fs::read_to_string("/proc/sys/kernel/pid_max")
|
||||
.expect("读取 pid_max 失败")
|
||||
.trim()
|
||||
.parse::<u32>()
|
||||
.expect("解析 pid_max 失败");
|
||||
pid_max + 1
|
||||
}
|
||||
|
||||
/// 陈旧身份：pid 可控，其余字段为占位值。
|
||||
fn stale_identity(pid: u32, socket_path: &Path) -> ServerIdentity {
|
||||
ServerIdentity {
|
||||
pid,
|
||||
socket_path: socket_path.to_str().unwrap().to_string(),
|
||||
version: "0.0.0".into(),
|
||||
started_at: 0,
|
||||
starttime: 0,
|
||||
}
|
||||
}
|
||||
|
||||
/// 写读回环：写入文件后读回五字段逐字段一致。
|
||||
#[test]
|
||||
fn identity_roundtrip() {
|
||||
let dir = TestDir::new();
|
||||
let path = dir.path("sidecar.json");
|
||||
let id = ServerIdentity::new("/tmp/oxide-test.sock");
|
||||
write_exclusive(&id, &path).expect("写 sidecar 应成功");
|
||||
let read = read_identity(&path).expect("读回 sidecar 应成功");
|
||||
assert_eq!(read, id, "五字段应逐字段一致");
|
||||
}
|
||||
|
||||
/// O_EXCL 排他：同路径二次 create_new 必得 AlreadyExists。
|
||||
#[test]
|
||||
fn write_exclusive_second_create_fails() {
|
||||
let dir = TestDir::new();
|
||||
let path = dir.path("sidecar.json");
|
||||
let id = ServerIdentity::new("/tmp/oxide-test.sock");
|
||||
write_exclusive(&id, &path).expect("首次写入应成功");
|
||||
let err = write_exclusive(&id, &path).unwrap_err();
|
||||
assert_eq!(err.kind(), io::ErrorKind::AlreadyExists, "二次 create_new 必得 AlreadyExists");
|
||||
}
|
||||
|
||||
/// 认领分支：无 sidecar 时注册成功，sidecar 是本进程身份。
|
||||
#[test]
|
||||
fn claim_no_sidecar_registers() {
|
||||
let dir = TestDir::new();
|
||||
let sidecar = dir.path("sidecar.json");
|
||||
let socket = dir.path("server.sock");
|
||||
assert_eq!(claim_sidecar(&sidecar, &socket), ClaimResult::Registered);
|
||||
let id = read_identity(&sidecar).expect("sidecar 应已写入");
|
||||
assert_eq!(id.pid, std::process::id());
|
||||
assert_eq!(id.socket_path, socket.to_str().unwrap());
|
||||
}
|
||||
|
||||
/// 认领分支：存活 server（真实监听者绑 socket 路径）拒绝注册。
|
||||
#[test]
|
||||
fn claim_live_server_refused() {
|
||||
let dir = TestDir::new();
|
||||
let sidecar = dir.path("sidecar.json");
|
||||
let socket = dir.path("server.sock");
|
||||
// 存活 server 必有 sidecar（启动顺序契约：先写 sidecar 再 bind socket）。
|
||||
let listener = UnixListener::bind(&socket).expect("绑定监听者应成功");
|
||||
write_exclusive(&ServerIdentity::new(socket.to_str().unwrap()), &sidecar).expect("写 sidecar 应成功");
|
||||
assert_eq!(claim_sidecar(&sidecar, &socket), ClaimResult::AlreadyRunning);
|
||||
drop(listener);
|
||||
}
|
||||
|
||||
/// 认领分支：死 PID 无 socket 判陈旧，sidecar 重建为本进程身份。
|
||||
#[test]
|
||||
fn claim_dead_pid_stale_rebuilt() {
|
||||
let dir = TestDir::new();
|
||||
let sidecar = dir.path("sidecar.json");
|
||||
let socket = dir.path("server.sock");
|
||||
write_exclusive(&stale_identity(dead_pid(), &socket), &sidecar).expect("写陈旧 sidecar 应成功");
|
||||
assert_eq!(claim_sidecar(&sidecar, &socket), ClaimResult::Registered);
|
||||
let id = read_identity(&sidecar).expect("重建 sidecar 应可读");
|
||||
assert_eq!(id.pid, std::process::id(), "重建 sidecar 应为本进程身份");
|
||||
}
|
||||
|
||||
/// 认领分支：存活 PID 加失配启动时刻判 PID 复用，sidecar 重建。
|
||||
#[test]
|
||||
fn claim_reused_pid_stale_rebuilt() {
|
||||
let dir = TestDir::new();
|
||||
let sidecar = dir.path("sidecar.json");
|
||||
let socket = dir.path("server.sock");
|
||||
// PID 是本进程（存活），启动时刻刻度为占位值：PID 已复用。
|
||||
write_exclusive(&stale_identity(std::process::id(), &socket), &sidecar).expect("写陈旧 sidecar 应成功");
|
||||
assert_eq!(claim_sidecar(&sidecar, &socket), ClaimResult::Registered);
|
||||
}
|
||||
|
||||
/// 认领分支：存活 PID 加匹配启动时刻保守拒绝。
|
||||
#[test]
|
||||
fn claim_alive_pid_matching_starttime_refused() {
|
||||
let dir = TestDir::new();
|
||||
let sidecar = dir.path("sidecar.json");
|
||||
let socket = dir.path("server.sock");
|
||||
// PID 与启动时刻刻度都是本进程的：进程活且刻度匹配。
|
||||
let id = ServerIdentity::new(socket.to_str().unwrap());
|
||||
write_exclusive(&id, &sidecar).expect("写 sidecar 应成功");
|
||||
assert_eq!(claim_sidecar(&sidecar, &socket), ClaimResult::RefusedAmbiguous);
|
||||
}
|
||||
|
||||
/// 认领分支：损坏文件加死 socket 判陈旧，sidecar 重建。
|
||||
#[test]
|
||||
fn claim_corrupted_sidecar_stale_rebuilt() {
|
||||
let dir = TestDir::new();
|
||||
let sidecar = dir.path("sidecar.json");
|
||||
let socket = dir.path("server.sock");
|
||||
// 损坏文件：半写 JSON。
|
||||
fs::write(&sidecar, r#"{"pid":1,"socket_path":""#).expect("写损坏文件应成功");
|
||||
assert_eq!(claim_sidecar(&sidecar, &socket), ClaimResult::Registered);
|
||||
let id = read_identity(&sidecar).expect("重建 sidecar 应可读");
|
||||
assert_eq!(id.pid, std::process::id());
|
||||
}
|
||||
|
||||
/// 认领分支：损坏文件加存活 socket 拒绝（socket 是存活事实源）。
|
||||
#[test]
|
||||
fn claim_corrupted_sidecar_live_socket_refused() {
|
||||
let dir = TestDir::new();
|
||||
let sidecar = dir.path("sidecar.json");
|
||||
let socket = dir.path("server.sock");
|
||||
let listener = UnixListener::bind(&socket).expect("绑定监听者应成功");
|
||||
fs::write(&sidecar, "not json at all").expect("写损坏文件应成功");
|
||||
assert_eq!(claim_sidecar(&sidecar, &socket), ClaimResult::AlreadyRunning);
|
||||
drop(listener);
|
||||
}
|
||||
|
||||
/// --rm 路径：两次调用路径不同，且文件名含本进程 PID 与 oxide-rm- 前缀。
|
||||
#[test]
|
||||
fn rm_socket_path_unique_and_contains_pid() {
|
||||
let a = rm_socket_path();
|
||||
std::thread::sleep(Duration::from_nanos(1000));
|
||||
let b = rm_socket_path();
|
||||
assert_ne!(a, b, "两次调用应得不同路径");
|
||||
let file_name = a.file_name().unwrap().to_string_lossy().into_owned();
|
||||
assert!(file_name.contains(&std::process::id().to_string()), "文件名应含本进程 PID");
|
||||
assert!(file_name.starts_with("oxide-rm-"), "文件名应以 oxide-rm- 前缀开头");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user