diff --git a/Cargo.lock b/Cargo.lock index 118ced0c..4b742b03 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.15.13" +version = "0.15.14" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.15.13" +version = "0.15.14" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.15.13" +version = "0.15.14" dependencies = [ "ansi_term", "bumpalo", @@ -1228,7 +1228,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.15.13" +version = "0.15.14" dependencies = [ "lru", "oxide_bytecode", @@ -1237,7 +1237,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1257,7 +1257,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1271,7 +1271,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.15.13" +version = "0.15.14" dependencies = [ "num-bigint", "oxc_allocator", @@ -1286,7 +1286,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1298,7 +1298,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1307,7 +1307,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.15.13" +version = "0.15.14" dependencies = [ "bumpalo", "dashmap", @@ -1321,7 +1321,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1333,7 +1333,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.15.13" +version = "0.15.14" dependencies = [ "tracing", "tracing-appender", @@ -1342,11 +1342,11 @@ dependencies = [ [[package]] name = "oxide_net" -version = "0.15.13" +version = "0.15.14" [[package]] name = "oxide_parser" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1377,7 +1377,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.15.13" +version = "0.15.14" dependencies = [ "num-bigint", "num-traits", @@ -1390,7 +1390,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.15.13" +version = "0.15.14" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1408,7 +1408,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.15.13" +version = "0.15.14" dependencies = [ "bumpalo", "dashmap", @@ -1421,7 +1421,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.15.13" +version = "0.15.14" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 8f837258..57020aeb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.15.13" +version = "0.15.14" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_cli/src/server/client.rs b/crates/oxide_cli/src/server/client.rs index 6f851198..e7ab0dee 100644 --- a/crates/oxide_cli/src/server/client.rs +++ b/crates/oxide_cli/src/server/client.rs @@ -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 Result { // 读 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 { + // 连接：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)); diff --git a/crates/oxide_cli/src/server/server.rs b/crates/oxide_cli/src/server/server.rs index c40a4b19..352b05db 100644 --- a/crates/oxide_cli/src/server/server.rs +++ b/crates/oxide_cli/src/server/server.rs @@ -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); } diff --git a/crates/oxide_cli/src/server/spawn.rs b/crates/oxide_cli/src/server/spawn.rs index fae42d19..3933f85b 100644 --- a/crates/oxide_cli/src/server/spawn.rs +++ b/crates/oxide_cli/src/server/spawn.rs @@ -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, 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, 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}"); + } +} diff --git a/crates/oxide_cli/tests/server_client.rs b/crates/oxide_cli/tests/server_client.rs index 7752043e..b2edf5ac 100644 --- a/crates/oxide_cli/tests/server_client.rs +++ b/crates/oxide_cli/tests/server_client.rs @@ -73,12 +73,16 @@ fn control_client_end_to_end() { other => panic!("应得 Status 帧，实得 {other:?}"), } - // 健康请求：healthy 为真。 - let response = send_control_request_from(&config.sidecar_path, &ServerRequest::Health).expect("健康请求应成功"); - assert!( - matches!(response, ServerResponse::Health { healthy: true }), - "healthy 应为真：{response:?}" - ); + // 健康请求：池预热异步，轮询至 healthy 为真（预热完成）。 + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let response = send_control_request_from(&config.sidecar_path, &ServerRequest::Health).expect("健康请求应成功"); + if matches!(response, ServerResponse::Health { healthy: true }) { + break; + } + assert!(Instant::now() < deadline, "worker 池预热应完成"); + std::thread::sleep(Duration::from_millis(10)); + } // 信息请求：socket 路径与配置一致、监听地址与配置一致、进程号为当前进程号。 let response = send_control_request_from(&config.sidecar_path, &ServerRequest::Info).expect("信息请求应成功"); @@ -144,12 +148,16 @@ fn control_client_tcp_arm_end_to_end() { other => panic!("应得 Version 帧，实得 {other:?}"), } - // 健康请求：healthy 为真。 - let response = send_control_request_from(&config.sidecar_path, &ServerRequest::Health).expect("健康请求应成功"); - assert!( - matches!(response, ServerResponse::Health { healthy: true }), - "healthy 应为真：{response:?}" - ); + // 健康请求：池预热异步，轮询至 healthy 为真（预热完成）。 + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let response = send_control_request_from(&config.sidecar_path, &ServerRequest::Health).expect("健康请求应成功"); + if matches!(response, ServerResponse::Health { healthy: true }) { + break; + } + assert!(Instant::now() < deadline, "worker 池预热应完成"); + std::thread::sleep(Duration::from_millis(10)); + } // 信息请求：endpoint 值与配置一致（host:port 文本形态）、监听地址与配置一致、 // 进程号为当前进程号。