mirror of
https://github.com/pjh456/OxideJS.git
synced 2026-10-23 18:21:13 +08:00
feat: 就绪探测抽端点级入口与远程变体,健康响应反映真实池状态
This commit is contained in:
@@ -1,7 +1,7 @@
|
||||
//! 控制客户端:读 sidecar、连接 server、发一帧控制请求、读回一帧响应。
|
||||
//!
|
||||
//! 关键约定:
|
||||
//! - 连接超时 5 秒(安全网):本地 Unix socket 连接立即完成或立即失败,
|
||||
//! - 连接超时 5 秒(安全网):本地连接立即完成或立即失败,
|
||||
//! 超时分支只覆盖 server 挂起等异常情形。
|
||||
//! - 读超时 5 秒(与交接路径的读超时同值):server 不回应即判连接失败,
|
||||
//! 不挂死。
|
||||
@@ -52,25 +52,19 @@ pub fn send_control_request(request: &ServerRequest) -> Result<ServerResponse, C
|
||||
send_control_request_from(&sidecar::well_known_sidecar_path(), request)
|
||||
}
|
||||
|
||||
/// 注入 sidecar 路径的入口(测试与复用):读 sidecar 取 socket 路径与 token,
|
||||
/// 连接、发认证帧、读 AuthOk、发请求、读响应,三分支错误映射。
|
||||
/// 注入 sidecar 路径的入口(测试与复用):读 sidecar 取 endpoint 与 token,
|
||||
/// 转调端点级入口。
|
||||
///
|
||||
/// # 步骤
|
||||
/// 1. 读 sidecar;缺失或损坏(解析失败)同归连接失败分支。
|
||||
/// 2. 连接 socket(5 秒超时安全网)。
|
||||
/// 3. 设读超时与写超时各 5 秒。
|
||||
/// 4. 发认证帧(token 取 sidecar),读回 `AuthOk`;非 `AuthOk` 判 server
|
||||
/// 错误(server 版本过旧需重启)。
|
||||
/// 5. 写请求帧。
|
||||
/// 6. 读响应帧;io 错误或 server 未应答即关闭(EOF)判连接失败。
|
||||
/// 7. 解析响应帧;畸形判协议错误,Error 帧判 server 错误,其余返回。
|
||||
/// 2. 取 sidecar 记录的 endpoint 文本形态(host:port 或路径)与 token,
|
||||
/// 转调 `send_control_request_at`。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - sidecar 路径由调用方注入,生产入口用 well-known 路径。
|
||||
/// - 响应帧为 Error 变体时不视为成功,message 原样带出。
|
||||
///
|
||||
/// # 副作用
|
||||
/// - 建立并关闭一条 Unix socket 连接。
|
||||
/// - 建立并关闭一条连接(Unix socket 或 TCP)。
|
||||
pub fn send_control_request_from(sidecar_path: &Path, request: &ServerRequest) -> Result<ServerResponse, ClientError> {
|
||||
// 读 sidecar:缺失或损坏都判连接失败,消息提示先启动 server。
|
||||
let identity = sidecar::read_identity(sidecar_path).ok_or_else(|| {
|
||||
@@ -80,9 +74,31 @@ pub fn send_control_request_from(sidecar_path: &Path, request: &ServerRequest) -
|
||||
))
|
||||
})?;
|
||||
|
||||
// 连接:5 秒超时是安全网,本地连接立即完成或立即失败。endpoint 取 sidecar
|
||||
// 记录的文本形态(host:port 或路径),撤掉 Unix 过渡包装。
|
||||
let stream = connect_with_timeout(&sidecar::parse_endpoint(&identity.socket_path), CONNECT_TIMEOUT)?;
|
||||
// endpoint 取 sidecar 记录的文本形态(host:port 或路径),token 取 sidecar。
|
||||
send_control_request_at(&sidecar::parse_endpoint(&identity.socket_path), &identity.token, request)
|
||||
}
|
||||
|
||||
/// 端点级控制请求入口:连接给定 endpoint、发认证帧(token 由调用方注入)、
|
||||
/// 发请求帧、读响应帧,三分支错误映射。
|
||||
///
|
||||
/// # 步骤
|
||||
/// 1. 连接 endpoint(5 秒超时安全网)。
|
||||
/// 2. 设读超时与写超时各 5 秒。
|
||||
/// 3. 发认证帧(token 由调用方注入),读回 `AuthOk`;非 `AuthOk` 判 server
|
||||
/// 错误(server 版本过旧需重启)。
|
||||
/// 4. 写请求帧。
|
||||
/// 5. 读响应帧;io 错误或 server 未应答即关闭(EOF)判连接失败。
|
||||
/// 6. 解析响应帧;畸形判协议错误,Error 帧判 server 错误,其余返回。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - endpoint 与 token 均由调用方注入:远程场景的 token 由拉起方传入。
|
||||
/// - 响应帧为 Error 变体时不视为成功,message 原样带出。
|
||||
///
|
||||
/// # 副作用
|
||||
/// - 建立并关闭一条连接(Unix socket 或 TCP)。
|
||||
pub fn send_control_request_at(endpoint: &Endpoint, token: &str, request: &ServerRequest) -> Result<ServerResponse, ClientError> {
|
||||
// 连接:5 秒超时是安全网,本地连接立即完成或立即失败。
|
||||
let stream = connect_with_timeout(endpoint, CONNECT_TIMEOUT)?;
|
||||
|
||||
// 读写超时各 5 秒:server 挂起(不回应)时不无限阻塞。
|
||||
let _ = stream.set_read_timeout(Some(CONNECT_TIMEOUT));
|
||||
@@ -93,9 +109,9 @@ pub fn send_control_request_from(sidecar_path: &Path, request: &ServerRequest) -
|
||||
.try_clone()
|
||||
.map_err(|e| ClientError::Connect(format!("克隆流失败:{e}")))?;
|
||||
|
||||
// 发认证帧:token 取 sidecar,server 侧逐字节异或判等。
|
||||
// 发认证帧:token 由调用方注入,server 侧逐字节异或判等。
|
||||
writer
|
||||
.write_all(protocol::encode_request(&ServerRequest::Auth { token: identity.token }).as_bytes())
|
||||
.write_all(protocol::encode_request(&ServerRequest::Auth { token: token.to_string() }).as_bytes())
|
||||
.map_err(|e| ClientError::Connect(format!("写认证帧失败:{e}")))?;
|
||||
|
||||
let mut reader = FrameReader::new(BufReader::new(stream));
|
||||
|
||||
@@ -903,7 +903,8 @@ fn dispatch_forge(
|
||||
|
||||
/// 控制请求分派:六类控制请求直接应答,不占 worker。
|
||||
///
|
||||
/// Status 响应读全体 worker 池的共享聚合计数器。
|
||||
/// Status 响应读全体 worker 池的共享聚合计数器;Health 响应反映真实健康
|
||||
/// (池预热完成加未进入关闭序列),不是「活着即健康」。
|
||||
fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerResponse {
|
||||
match request {
|
||||
ServerRequest::Version => ServerResponse::Version {
|
||||
@@ -915,7 +916,11 @@ fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerRespo
|
||||
pool_peak: ctx.pool_counters.peak(),
|
||||
uptime_ms: ctx.started_at.elapsed().as_millis() as u64,
|
||||
},
|
||||
ServerRequest::Health => ServerResponse::Health { healthy: true },
|
||||
// 真实健康:池预热完成(全体 worker 池聚合计数不小于 1)加未进入
|
||||
// 关闭序列(排水进行中不再健康)。
|
||||
ServerRequest::Health => ServerResponse::Health {
|
||||
healthy: !ctx.shutdown.load(Ordering::SeqCst) && ctx.pool_counters.total() >= 1,
|
||||
},
|
||||
ServerRequest::Info => ServerResponse::Info {
|
||||
version: env!("CARGO_PKG_VERSION").to_string(),
|
||||
socket_path: ctx.socket_path.clone(),
|
||||
@@ -1175,17 +1180,23 @@ mod tests {
|
||||
stop_server(&mut stream, handle);
|
||||
}
|
||||
|
||||
/// 健康请求得 Health 帧(骨架阶段活着即健康)。
|
||||
/// 健康请求得 Health 帧:池预热完成加未进入关闭序列时 healthy 为真。
|
||||
#[test]
|
||||
fn health_response() {
|
||||
let config = unique_paths("health");
|
||||
let handle = start_server(config.clone());
|
||||
let mut stream = connect_auth(&config);
|
||||
let response = send_and_recv(&mut stream, &ServerRequest::Health);
|
||||
assert!(
|
||||
matches!(response, ServerResponse::Health { healthy: true }),
|
||||
"healthy 应为真:{response:?}"
|
||||
);
|
||||
|
||||
// worker 池异步预热:轮询至 healthy 为真(预热完成)。
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
loop {
|
||||
let response = send_and_recv(&mut stream, &ServerRequest::Health);
|
||||
if matches!(response, ServerResponse::Health { healthy: true }) {
|
||||
break;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "worker 池预热应完成");
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
stop_server(&mut stream, handle);
|
||||
}
|
||||
|
||||
@@ -1239,12 +1250,17 @@ mod tests {
|
||||
let response = protocol::parse_response(&frame).expect("响应帧解析应成功");
|
||||
assert!(matches!(response, ServerResponse::Error { .. }), "应得 Error 帧:{response:?}");
|
||||
|
||||
// 连接保持可用:后续健康请求正常应答。
|
||||
let response = send_and_recv(&mut stream, &ServerRequest::Health);
|
||||
assert!(
|
||||
matches!(response, ServerResponse::Health { healthy: true }),
|
||||
"连接应保持可用:{response:?}"
|
||||
);
|
||||
// 连接保持可用:后续健康请求正常应答(池预热异步,轮询至 healthy
|
||||
// 为真)。
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
loop {
|
||||
let response = send_and_recv(&mut stream, &ServerRequest::Health);
|
||||
if matches!(response, ServerResponse::Health { healthy: true }) {
|
||||
break;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "worker 池预热应完成");
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
stop_server(&mut stream, handle);
|
||||
}
|
||||
|
||||
@@ -1739,11 +1755,16 @@ mod tests {
|
||||
let handle = start_rm_server(config.clone());
|
||||
let mut stream = Conn::Unix(UnixStream::connect(unix_path(&config.endpoint)).expect("首个连接应成功"));
|
||||
authenticate(&mut stream, &config.token);
|
||||
let response = send_and_recv(&mut stream, &ServerRequest::Health);
|
||||
assert!(
|
||||
matches!(response, ServerResponse::Health { healthy: true }),
|
||||
"应得 Health 帧:{response:?}"
|
||||
);
|
||||
// 池预热异步:轮询至 healthy 为真(预热完成)。
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
loop {
|
||||
let response = send_and_recv(&mut stream, &ServerRequest::Health);
|
||||
if matches!(response, ServerResponse::Health { healthy: true }) {
|
||||
break;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "worker 池预热应完成");
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
|
||||
// 监听器已丢弃:第二个连接得连接被拒绝。轮询至拒绝,容忍连接线程
|
||||
// 建立与监听器丢弃之间的极小时序窗口(并发套件下调度延迟放大窗口)。
|
||||
@@ -1814,11 +1835,16 @@ mod tests {
|
||||
let handle = start_server(config.clone());
|
||||
let mut stream = connect(&config);
|
||||
authenticate(&mut stream, &read_token(&config));
|
||||
let response = send_and_recv(&mut stream, &ServerRequest::Health);
|
||||
assert!(
|
||||
matches!(response, ServerResponse::Health { healthy: true }),
|
||||
"认证后应正常应答:{response:?}"
|
||||
);
|
||||
// 池预热异步:轮询至 healthy 为真(预热完成)。
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
loop {
|
||||
let response = send_and_recv(&mut stream, &ServerRequest::Health);
|
||||
if matches!(response, ServerResponse::Health { healthy: true }) {
|
||||
break;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "worker 池预热应完成");
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
}
|
||||
stop_server(&mut stream, handle);
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! spawn 与就绪探测:拉起脱离 server 与就绪探测的唯一共享入口。
|
||||
//!
|
||||
//! start、restart、watchdog 三条路径共用本模块的同一对函数,不另起第二套
|
||||
//! start、restart、watchdog 三条路径共用本模块的同一组函数,不另起第二套
|
||||
//! spawn 逻辑。`current_exe()` 在 lib 上下文与二进制目标返回同一路径(lib
|
||||
//! 链接进同一个 oxide 二进制),spawn 行为逐字节不变。
|
||||
|
||||
@@ -8,7 +8,9 @@ use std::net::IpAddr;
|
||||
use std::process::{Command, Stdio};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use super::client::send_control_request;
|
||||
use oxide_net::Endpoint;
|
||||
|
||||
use super::client::{send_control_request, send_control_request_at};
|
||||
use super::protocol::{ServerRequest, ServerResponse};
|
||||
|
||||
/// spawn 脱离的前台 server 子进程(`server start --foreground --host H --port N`):
|
||||
@@ -48,7 +50,7 @@ pub fn spawn_detached_server(workers: Option<u32>, host: IpAddr, port: u16) -> s
|
||||
/// 就绪探测:轮询健康请求直至得 healthy。
|
||||
///
|
||||
/// # 步骤
|
||||
/// 1. 循环发健康请求,得 healthy 即返回成功。
|
||||
/// 1. 循环发健康请求(well-known sidecar 入口),得 healthy 即返回成功。
|
||||
/// 2. 超过截止返回含超时秒数的消息。
|
||||
///
|
||||
/// # 边界与前提
|
||||
@@ -59,11 +61,42 @@ pub fn spawn_detached_server(workers: Option<u32>, host: IpAddr, port: u16) -> s
|
||||
/// # 副作用
|
||||
/// - 每轮建立并关闭一条 Unix socket 连接。
|
||||
pub fn wait_server_ready(deadline: Duration) -> Result<(), String> {
|
||||
wait_ready_until(
|
||||
|| matches!(send_control_request(&ServerRequest::Health), Ok(ServerResponse::Health { healthy: true })),
|
||||
deadline,
|
||||
)
|
||||
}
|
||||
|
||||
/// 远程就绪探测:对着给定 endpoint(带 token)轮询健康请求直至得 healthy。
|
||||
///
|
||||
/// # 步骤
|
||||
/// 1. 循环对 endpoint 发健康请求(token 由调用方注入),得 healthy 即返回成功。
|
||||
/// 2. 超过截止返回含超时秒数的消息。
|
||||
///
|
||||
/// # 边界与前提
|
||||
/// - endpoint 与 token 均由调用方注入:远程场景的 token 由拉起方传入,
|
||||
/// 截止时长由调用方按场景注入。
|
||||
/// - 连接与读写超时沿用控制客户端 5 秒口径,轮询间隔 100 毫秒,与
|
||||
/// well-known 入口同口径。
|
||||
///
|
||||
/// # 副作用
|
||||
/// - 每轮建立并关闭一条连接(Unix socket 或 TCP)。
|
||||
pub fn wait_server_ready_remote(endpoint: &Endpoint, token: &str, deadline: Duration) -> Result<(), String> {
|
||||
wait_ready_until(
|
||||
|| matches!(
|
||||
send_control_request_at(endpoint, token, &ServerRequest::Health),
|
||||
Ok(ServerResponse::Health { healthy: true })
|
||||
),
|
||||
deadline,
|
||||
)
|
||||
}
|
||||
|
||||
/// 共享探测循环:反复调用探测函数,100 毫秒间隔,探测返回真即成功;
|
||||
/// 超过截止返回含超时秒数的消息。
|
||||
fn wait_ready_until(probe: impl Fn() -> bool, 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 {
|
||||
if probe() {
|
||||
return Ok(());
|
||||
}
|
||||
if Instant::now() >= end {
|
||||
@@ -72,3 +105,52 @@ pub fn wait_server_ready(deadline: Duration) -> Result<(), String> {
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io::{BufReader, Write};
|
||||
use std::net::{SocketAddr, TcpListener};
|
||||
|
||||
use crate::server::protocol::{self, FrameReader};
|
||||
|
||||
/// 假 TCP server:绑动态端口,接受循环逐连接读认证帧回 AuthOk、
|
||||
/// 读请求帧回给定的健康帧。
|
||||
fn fake_tcp_health_server(healthy: bool) -> SocketAddr {
|
||||
let listener = TcpListener::bind("127.0.0.1:0").expect("绑定假 server 应成功");
|
||||
let addr = listener.local_addr().expect("取端口应成功");
|
||||
std::thread::spawn(move || {
|
||||
while let Ok((mut stream, _)) = listener.accept() {
|
||||
let mut reader = FrameReader::new(BufReader::new(stream.try_clone().expect("克隆流应成功")));
|
||||
let _ = reader.read_frame();
|
||||
let auth_ok = protocol::encode_response(&ServerResponse::AuthOk);
|
||||
if stream.write_all(auth_ok.as_bytes()).is_err() {
|
||||
break;
|
||||
}
|
||||
let _ = reader.read_frame();
|
||||
let health = protocol::encode_response(&ServerResponse::Health { healthy });
|
||||
if stream.write_all(health.as_bytes()).is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
addr
|
||||
}
|
||||
|
||||
/// TCP 臂:假 server 回 AuthOk 加 healthy 健康帧,就绪探测返回成功。
|
||||
#[test]
|
||||
fn remote_ready_tcp_arm() {
|
||||
let addr = fake_tcp_health_server(true);
|
||||
let result = wait_server_ready_remote(&Endpoint::Tcp(addr), "test-token", Duration::from_secs(5));
|
||||
assert!(result.is_ok(), "就绪探测应成功:{result:?}");
|
||||
}
|
||||
|
||||
/// TCP 臂超时:假 server 恒回不健康帧,就绪探测在截止后返回超时消息。
|
||||
#[test]
|
||||
fn remote_ready_timeout_when_unhealthy() {
|
||||
let addr = fake_tcp_health_server(false);
|
||||
let result = wait_server_ready_remote(&Endpoint::Tcp(addr), "test-token", Duration::from_millis(300));
|
||||
let err = result.expect_err("应判超时");
|
||||
assert!(err.contains("就绪"), "应得超时消息:{err}");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user