feat: 客户端与探活三函数按 endpoint 形态分支 Unix 与 TCP 两臂

This commit is contained in:
2026-10-11 13:07:29 +00:00
parent d6fa71759b
commit 9d2790e036
9 changed files with 149 additions and 78 deletions
+3 -1
View File
@@ -23,6 +23,8 @@ use std::path::Path;
use std::process::Command;
use std::time::{Duration, Instant};
use oxide_net::Endpoint;
use super::sidecar::{self, ServerIdentity};
/// 缺省杀进程等待:10 秒,长于 server 的在途排水安全超时 5 秒,正常优雅
@@ -151,7 +153,7 @@ pub fn cleanup_with_wait(sidecar_path: &Path, socket_path: &Path, kill_wait: Dur
// PID 存活与刻度匹配(PID 死时匹配位恒假)。
let identity = sidecar::read_identity(sidecar_path);
let socket_exists = socket_path.exists();
let socket_alive = socket_exists && sidecar::is_server_alive(socket_path);
let socket_alive = socket_exists && sidecar::is_server_alive(&Endpoint::Unix(socket_path.to_path_buf()));
let (pid_alive, starttime_match) = match &identity {
Some(id) => {
let alive = sidecar::pid_alive(id.pid);
+19 -11
View File
@@ -8,12 +8,15 @@
//! - 错误三分支:连接失败、协议错误、server 错误帧,各映射到可读消息,
//! 不 panic。
use std::io::{BufReader, Write};
use std::io::BufReader;
use std::net::TcpStream;
use std::os::unix::net::UnixStream;
use std::path::Path;
use std::path::{Path, PathBuf};
use std::sync::mpsc::RecvTimeoutError;
use std::time::Duration;
use oxide_net::{Conn, Endpoint};
use super::protocol::{self, FrameReader, ProtocolError, ServerRequest, ServerResponse};
use super::sidecar;
@@ -76,7 +79,7 @@ pub fn send_control_request_from(sidecar_path: &Path, request: &ServerRequest) -
})?;
// 连接:5 秒超时是安全网,本地 socket 连接立即完成或立即失败。
let mut stream = connect_with_timeout(Path::new(&identity.socket_path), CONNECT_TIMEOUT)?;
let mut stream = connect_with_timeout(&Endpoint::Unix(PathBuf::from(identity.socket_path)), CONNECT_TIMEOUT)?;
// 读写超时各 5 秒:server 挂起(不回应)时不无限阻塞。
let _ = stream.set_read_timeout(Some(CONNECT_TIMEOUT));
@@ -102,29 +105,33 @@ pub fn send_control_request_from(sidecar_path: &Path, request: &ServerRequest) -
}
}
/// 带超时的连接:线程内连接、通道上带超时取结果。
/// 带超时的连接:线程内连接、通道上带超时取结果,两臂按 endpoint 形态分支。
///
/// # 步骤
/// 1. 派生线程连接 socket。
/// 1. 派生线程连接 endpoint(Unix 路径或 TCP 地址)。
/// 2. 通道上带超时取结果;超时判连接失败。
///
/// # 边界与前提
/// - 超时后连接线程继续存活到连接完成或失败(本地 socket 立即出结果),
/// 发送端已被接收端丢弃,线程随即退出,无泄漏。
fn connect_with_timeout(path: &Path, timeout: Duration) -> Result<UnixStream, ClientError> {
fn connect_with_timeout(endpoint: &Endpoint, timeout: Duration) -> Result<Conn, ClientError> {
let (tx, rx) = std::sync::mpsc::channel();
let spawn_path = path.to_path_buf();
let display = path.display().to_string();
let spawn = endpoint.clone();
let display = endpoint.describe();
std::thread::spawn(move || {
let _ = tx.send(UnixStream::connect(&spawn_path));
let result = match &spawn {
Endpoint::Unix(path) => UnixStream::connect(path).map(Conn::Unix),
Endpoint::Tcp(addr) => TcpStream::connect(*addr).map(Conn::Tcp),
};
let _ = tx.send(result);
});
match rx.recv_timeout(timeout) {
Ok(Ok(stream)) => Ok(stream),
Ok(Err(e)) => Err(ClientError::Connect(format!(
"连接 server 失败:{e}(socket 路径 {display},可用 `oxide server cleanup` 清理陈旧文件)"
"连接 server 失败:{e}(endpoint {display},可用 `oxide server cleanup` 清理陈旧文件)"
))),
Err(RecvTimeoutError::Timeout) => Err(ClientError::Connect(format!(
"连接 server 超时({} 秒),socket 路径 {display}",
"连接 server 超时({} 秒),endpoint {display}",
timeout.as_secs()
))),
Err(RecvTimeoutError::Disconnected) => Err(ClientError::Connect("连接线程异常退出".into())),
@@ -135,6 +142,7 @@ fn connect_with_timeout(path: &Path, timeout: Duration) -> Result<UnixStream, Cl
mod tests {
use super::*;
use std::fs;
use std::io::Write;
use std::os::unix::net::UnixListener;
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};
+3 -1
View File
@@ -21,6 +21,8 @@
use std::path::Path;
use std::time::{Duration, Instant};
use oxide_net::Endpoint;
use super::cleanup::{self, Decision};
use super::sidecar;
@@ -92,7 +94,7 @@ pub fn liveness_scan_with_waits(
// PID 存活与刻度匹配(PID 死时匹配位恒假)。
let identity = sidecar::read_identity(sidecar_path);
let socket_exists = socket_path.exists();
let socket_alive = socket_exists && sidecar::is_server_alive(socket_path);
let socket_alive = socket_exists && sidecar::is_server_alive(&Endpoint::Unix(socket_path.to_path_buf()));
let (pid_alive, starttime_match) = match &identity {
Some(id) => {
let alive = sidecar::pid_alive(id.pid);
+75 -29
View File
@@ -21,8 +21,7 @@
use std::fs;
use std::io;
use std::io::Write;
use std::net::TcpListener;
use std::net::{TcpListener, TcpStream};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
@@ -456,19 +455,23 @@ fn finish_shutdown(
/// 交接接管:连接旧 server、发 yield 请求、等旧 server 排空退出后重新认领。
///
/// # 步骤
/// 1. 连接旧 server 的 socket;连接失败判 `YieldFailed`。
/// 1. 连接旧 server 的 endpoint(Unix 路径或 TCP 地址);连接失败判
/// `YieldFailed`。
/// 2. 设 5 秒读超时,写 yield 请求帧。
/// 3. 读一帧应答:读超时(旧 server 挂起)或应答非 yield 确认(旧版本
/// binary 无该变体时以错误帧答复未知类型)判 `YieldFailed`。
/// 4. 关闭连接,轮询 sidecar 文件消失(50 毫秒间隔、10 秒上限);超时判
/// `YieldFailed`。
/// 5. 重新认领 sidecar;得 `Registered` 即接管成功,其余态判 `YieldFailed`。
/// 4. 关闭连接,按 endpoint 形态等旧 server 退出:Unix 臂轮询 sidecar 文件
/// 消失,TCP 臂以端口释放(连接被拒)为退出信号;均 50 毫秒间隔、
/// 10 秒上限,超时判 `YieldFailed`。
/// 5. Unix 臂重新认领 sidecar;得 `Registered` 即接管成功,其余态判
/// `YieldFailed`。TCP 臂无 sidecar 文件,不重认领。
///
/// # 边界与前提
/// - 交接等待 10 秒长于旧 server 的排水安全超时 5 秒,正常路径必然先于
/// 等待上限完成。
/// - 旧 server 先删 socket 文件后删 sidecar,sidecar 消失时 socket 文件必
/// 已删除,接管绑定无竞态。
/// - Unix 臂:旧 server 先删 socket 文件后删 sidecar,sidecar 消失时
/// socket 文件必已删除,接管绑定无竞态。
/// - TCP 臂:监听者 drop 即端口释放,连接被拒即旧 server 已退出。
///
/// # 副作用
/// - 失败时不触碰旧 server 的文件(socket 与 sidecar),人工兜底归清理命令。
@@ -476,17 +479,18 @@ fn finish_shutdown(
/// # 注意事项
/// - 纯函数式入口:版本管理路径(版本不匹配强制交接)与显式交接路径调用
/// 同一入口。
/// - 交接协议在 Unix 文件面上;TCP 端点是防御性分支,不参与交接。
/// - 退出信号源按形态分:Unix 臂是 sidecar 文件消失,TCP 臂是端口释放。
pub fn yield_and_takeover(config: &ServerConfig) -> Result<(), ServerError> {
let socket_path = match &config.endpoint {
Endpoint::Unix(path) => path,
Endpoint::Tcp(_) => return Err(ServerError::YieldFailed("TCP 端点不支持交接".into())),
// 连接旧 server:连接成功是交接的前提,两臂按 endpoint 形态分支。
let mut stream = match &config.endpoint {
Endpoint::Unix(path) => Conn::Unix(
UnixStream::connect(path).map_err(|e| ServerError::YieldFailed(format!("连接旧 server 失败:{e}")))?,
),
Endpoint::Tcp(addr) => Conn::Tcp(
TcpStream::connect(*addr).map_err(|e| ServerError::YieldFailed(format!("连接旧 server 失败:{e}")))?,
),
};
// 连接旧 server:连接成功是交接的前提。
let mut stream =
UnixStream::connect(socket_path).map_err(|e| ServerError::YieldFailed(format!("连接旧 server 失败:{e}")))?;
// 5 秒读超时:旧 server 挂起(不回应 yield)不无限阻塞。
let _ = stream.set_read_timeout(Some(Duration::from_secs(5)));
stream
@@ -504,22 +508,39 @@ pub fn yield_and_takeover(config: &ServerConfig) -> Result<(), ServerError> {
if !matches!(response, ServerResponse::Yield) {
return Err(ServerError::YieldFailed(format!("旧 server 未受理交接:{response:?}")));
}
// 关闭连接:旧 server 已受理,交接进入文件轮询阶段。
// 关闭连接:旧 server 已受理,交接进入退出等待阶段。
drop(reader);
// 轮询 sidecar 消失:旧 server 的退出序列以 sidecar 最后删除收口。
let deadline = Instant::now() + Duration::from_secs(10);
while config.sidecar_path.exists() {
if Instant::now() >= deadline {
return Err(ServerError::YieldFailed("旧 server 10 秒内未退出,sidecar 文件未消失".into()));
}
std::thread::sleep(Duration::from_millis(50));
}
match &config.endpoint {
// Unix 臂:轮询 sidecar 消失(旧 server 的退出序列以 sidecar 最后
// 删除收口),消失后重新认领。
Endpoint::Unix(path) => {
let deadline = Instant::now() + Duration::from_secs(10);
while config.sidecar_path.exists() {
if Instant::now() >= deadline {
return Err(ServerError::YieldFailed("旧 server 10 秒内未退出,sidecar 文件未消失".into()));
}
std::thread::sleep(Duration::from_millis(50));
}
// 重新认领:sidecar 已消失,O_EXCL 创建应成功;其余态判交接失败。
match sidecar::claim_sidecar(&config.sidecar_path, socket_path, &config.version) {
ClaimResult::Registered => Ok(()),
other => Err(ServerError::YieldFailed(format!("交接后重新认领未通过:{other:?}"))),
// 重新认领:sidecar 已消失,O_EXCL 创建应成功;其余态判交接失败。
match sidecar::claim_sidecar(&config.sidecar_path, path, &config.version) {
ClaimResult::Registered => Ok(()),
other => Err(ServerError::YieldFailed(format!("交接后重新认领未通过:{other:?}"))),
}
}
// TCP 臂:无 sidecar 文件可轮询,以端口释放(连接被拒)为旧 server
// 退出信号,10 秒上限,不重认领。
Endpoint::Tcp(addr) => {
let deadline = Instant::now() + Duration::from_secs(10);
while TcpStream::connect(*addr).is_ok() {
if Instant::now() >= deadline {
return Err(ServerError::YieldFailed("旧 server 10 秒内未退出,端口仍被占用".into()));
}
std::thread::sleep(Duration::from_millis(50));
}
Ok(())
}
}
}
@@ -812,6 +833,7 @@ fn wait_in_flight_drained(in_flight: &AtomicUsize, timeout: Duration) {
#[cfg(test)]
mod tests {
use super::*;
use std::io::Write;
use std::net::TcpStream;
use std::thread::JoinHandle;
@@ -1349,6 +1371,30 @@ mod tests {
assert!(unix_path(&config.endpoint).exists(), "模拟 socket 文件不应被删除");
}
/// 交接 TCP 臂:假 server 回 yield 确认帧后 drop 监听器(端口释放),
/// 交接以端口释放为退出信号返回成功(不重认领)。
#[test]
fn yield_takeover_tcp_arm() {
let listener = TcpListener::bind("127.0.0.1:0").expect("绑定假 server 应成功");
let addr = listener.local_addr().expect("取端口应成功");
std::thread::spawn(move || {
if let Ok((mut stream, _)) = listener.accept() {
let mut reader = FrameReader::new(io::BufReader::new(stream.try_clone().expect("克隆流应成功")));
let _ = reader.read_frame();
let frame = protocol::encode_response(&ServerResponse::Yield);
let _ = stream.write_all(frame.as_bytes());
}
});
let config = ServerConfig {
endpoint: Endpoint::Tcp(addr),
sidecar_path: std::env::temp_dir().join(format!("oxide_server_tcp_yield_{}_sidecar.json", std::process::id())),
worker_count: 2,
version: env!("CARGO_PKG_VERSION").to_string(),
};
yield_and_takeover(&config).expect("TCP 臂交接应成功");
}
/// --rm 加断开 EOF:eval 得完成值后客户端断开,server 在有界时间内
/// 正常退出,socket 文件被删除,全局 well-known 路径前后不变。
#[test]
+9 -4
View File
@@ -13,12 +13,14 @@
use std::fs::{self, File, OpenOptions};
use std::io::{self, Write};
use std::net::TcpStream;
use std::os::unix::io::AsRawFd;
use std::os::unix::net::UnixStream;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{SystemTime, UNIX_EPOCH};
use oxide_net::Endpoint;
use serde::{Deserialize, Serialize};
/// server 身份快照:单行 JSON sidecar 文件,启动时写入一次。
@@ -174,12 +176,15 @@ pub fn read_identity(path: &Path) -> Option<ServerIdentity> {
serde_json::from_str(&text).ok()
}
/// socket 探活:连接成功即存在存活 server。
/// endpoint 探活:连接成功即存在存活 server,两臂按形态分支。
///
/// 内核只把连接完成给存活监听者,该信号不依赖 PID、不受 PID 复用影响,
/// 是注册判定的唯一裁决。
pub fn is_server_alive(socket_path: &Path) -> bool {
UnixStream::connect(socket_path).is_ok()
pub fn is_server_alive(endpoint: &Endpoint) -> bool {
match endpoint {
Endpoint::Unix(path) => UnixStream::connect(path).is_ok(),
Endpoint::Tcp(addr) => TcpStream::connect(*addr).is_ok(),
}
}
/// PID 探活:`kill -0` 退出码 0 即进程存在。
@@ -254,7 +259,7 @@ pub fn claim_sidecar(sidecar_path: &Path, socket_path: &Path, new_version: &str)
}
// socket 探活是存活的唯一事实源:连接成功即有存活 server。
if is_server_alive(socket_path) {
if is_server_alive(&Endpoint::Unix(socket_path.to_path_buf())) {
// 读回旧 sidecar 与本进程版本比对:匹配拒绝启动,不匹配或不可读
// 交调用方强制交接。
match read_identity(sidecar_path) {