From aa40d96f2beb5f82352c7c40e79e8b6b82c6d8c4 Mon Sep 17 00:00:00 2001 From: pjh456 <147148383@qq.com> Date: Tue, 6 Oct 2026 16:31:02 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20server=20=E4=BA=A4=E6=8E=A5=20yield=20?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE=EF=BC=8C=E6=97=A7=20server=20=E6=8E=92?= =?UTF-8?q?=E7=A9=BA=E9=80=80=E5=87=BA=E5=90=8E=E6=96=B0=20server=20?= =?UTF-8?q?=E6=8E=A5=E7=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 36 ++-- Cargo.toml | 2 +- crates/oxide_cli/src/server/mod.rs | 3 +- crates/oxide_cli/src/server/protocol.rs | 14 +- crates/oxide_cli/src/server/server.rs | 253 +++++++++++++++++++++--- 5 files changed, 259 insertions(+), 49 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f215250a..a424975d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.8.31" +version = "0.8.32" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.8.31" +version = "0.8.32" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.8.31" +version = "0.8.32" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.8.31" +version = "0.8.32" dependencies = [ "ansi_term", "bumpalo", @@ -1225,7 +1225,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.8.31" +version = "0.8.32" dependencies = [ "lru", "oxide_bytecode", @@ -1234,7 +1234,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.8.31" +version = "0.8.32" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1253,7 +1253,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.8.31" +version = "0.8.32" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1267,7 +1267,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.8.31" +version = "0.8.32" dependencies = [ "num-bigint", "oxc_allocator", @@ -1282,7 +1282,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.8.31" +version = "0.8.32" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1291,7 +1291,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.8.31" +version = "0.8.32" dependencies = [ "bumpalo", "dashmap", @@ -1306,7 +1306,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.8.31" +version = "0.8.32" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1318,7 +1318,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.8.31" +version = "0.8.32" dependencies = [ "tracing", "tracing-appender", @@ -1327,7 +1327,7 @@ dependencies = [ [[package]] name = "oxide_parser" -version = "0.8.31" +version = "0.8.32" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1342,7 +1342,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.8.31" +version = "0.8.32" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1358,7 +1358,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.8.31" +version = "0.8.32" dependencies = [ "num-bigint", "num-traits", @@ -1370,7 +1370,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.8.31" +version = "0.8.32" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1387,7 +1387,7 @@ dependencies = [ [[package]] name = "oxide_types" -version = "0.8.31" +version = "0.8.32" dependencies = [ "bumpalo", "hashbrown 0.14.5", @@ -1399,7 +1399,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.8.31" +version = "0.8.32" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 96644afb..ade2070f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,7 +22,7 @@ members = [ ] [workspace.package] -version = "0.8.31" +version = "0.8.32" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_cli/src/server/mod.rs b/crates/oxide_cli/src/server/mod.rs index d8fb46b8..4a2d16a4 100644 --- a/crates/oxide_cli/src/server/mod.rs +++ b/crates/oxide_cli/src/server/mod.rs @@ -13,7 +13,8 @@ pub mod eval; /// 协议数据面：NDJSON 帧格式与 serde 结构体。 pub mod protocol; /// server 进程主体：accept 循环、控制请求分派、执行请求路由、信号处理 -/// （SIGINT/SIGTERM 置位关闭标志）与优雅退出。 +/// （SIGINT/SIGTERM 置位关闭标志）、交接协议（向旧 server 发 yield 请求使其 +/// 排空退出、新 server 接管 socket 路径）与优雅退出。 #[allow(clippy::module_inception)] pub mod server; /// 身份注册：sidecar 文件、liveness 检查与陈旧清理。 diff --git a/crates/oxide_cli/src/server/protocol.rs b/crates/oxide_cli/src/server/protocol.rs index a8fd516b..a1774f38 100644 --- a/crates/oxide_cli/src/server/protocol.rs +++ b/crates/oxide_cli/src/server/protocol.rs @@ -48,7 +48,7 @@ impl ProtocolError { } } -/// server 请求帧：eval 执行请求加五类控制请求。 +/// server 请求帧：eval 执行请求加六类控制请求。 /// /// 序列化为 tagged 对象，`type` 字段区分类型（snake_case）。 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -71,6 +71,8 @@ pub enum ServerRequest { Info, /// 优雅关闭。 Shutdown, + /// 交接请求：旧 server 排空在途后退出，让出 socket 路径。 + Yield, } /// server 响应帧：每类请求一个响应，外加协议错误帧。 @@ -103,6 +105,8 @@ pub enum ServerResponse { Info { version: String, socket_path: String, pid: u32 }, /// 关闭确认。 Shutdown, + /// 交接确认：旧 server 已受理，开始排空退出。 + Yield, /// 协议错误帧：畸形帧或未知类型，server 以该帧答复。 Error { message: String }, } @@ -126,10 +130,10 @@ impl ServerResponse { } /// 请求帧可取的 `type` 值（须与 `ServerRequest` 变体保持同步）。 -const REQUEST_TAGS: &[&str] = &["eval", "version", "status", "health", "info", "shutdown"]; +const REQUEST_TAGS: &[&str] = &["eval", "version", "status", "health", "info", "shutdown", "yield"]; /// 响应帧可取的 `type` 值（须与 `ServerResponse` 变体保持同步）。 -const RESPONSE_TAGS: &[&str] = &["eval_result", "version", "status", "health", "info", "shutdown", "error"]; +const RESPONSE_TAGS: &[&str] = &["eval_result", "version", "status", "health", "info", "shutdown", "yield", "error"]; /// 序列化请求为帧字符串（含结尾换行）。 /// @@ -311,7 +315,7 @@ mod tests { }); } - /// 五类控制请求 round-trip。 + /// 六类控制请求 round-trip。 #[test] fn control_request_roundtrip() { for req in [ @@ -320,6 +324,7 @@ mod tests { ServerRequest::Health, ServerRequest::Info, ServerRequest::Shutdown, + ServerRequest::Yield, ] { roundtrip_request(&req); } @@ -345,6 +350,7 @@ mod tests { pid: 1234, }, ServerResponse::Shutdown, + ServerResponse::Yield, ServerResponse::Error { message: "boom".into() }, ] { roundtrip_response(&resp); diff --git a/crates/oxide_cli/src/server/server.rs b/crates/oxide_cli/src/server/server.rs index 72555288..822887d4 100644 --- a/crates/oxide_cli/src/server/server.rs +++ b/crates/oxide_cli/src/server/server.rs @@ -6,10 +6,11 @@ //! - accept 循环为非阻塞 accept 加 10 毫秒轮询加原子关闭标志：唤醒路径 //! 确定、无锁、无唤醒丢失。 //! - 并发模型：主线程只跑 accept 循环；每连接一个标准库线程只做 I/O -//! （帧读写）；五类控制请求由连接线程直接应答、不占 worker；执行请求 +//! （帧读写）；六类控制请求由连接线程直接应答、不占 worker；执行请求 //! 经 mpsc 路由到固定 N 个常驻 worker 执行（各持自己线程上的 VM 池）。 -//! - 优雅退出顺序：关闭请求或 SIGINT/SIGTERM 信号置位关闭标志 → 在途归零 -//! → 清空发送端 → join worker → drop 内核 → 删 sidecar 与 socket 文件。 +//! - 优雅退出顺序：关闭请求、SIGINT/SIGTERM 信号或 yield 请求置位关闭标志 +//! → 在途归零 → 清空发送端 → join worker → drop 内核 → 先删 socket 文件 +//! 后删 sidecar（新 server 以 sidecar 消失为接管信号）。 //! - 池本体不是 Send/Sync（Vm 不跨线程），上下文只持池的原子状态句柄； //! 状态响应读全体 worker 池的共享聚合计数器。 @@ -30,7 +31,7 @@ use super::sidecar::{self, ClaimResult}; use super::workers::{self, WorkerRouter}; /// server 启动与运行错误：绑定失败、认领被拒、接受连接失败、读取身份失败、 -/// 信号处理器注册失败。 +/// 信号处理器注册失败、交接失败。 #[derive(Debug, Clone, PartialEq, Eq)] pub enum ServerError { /// sidecar 认领未通过（已有存活 server 或存活证据不足以裁决）。 @@ -45,6 +46,8 @@ pub enum ServerError { WorkerPanic(String), /// 信号处理器注册失败（系统级错误，非"已注册"的幂等情形）。 SignalInstallFailed(String), + /// 交接失败（连不上旧 server、旧 server 不受理 yield、或交接后重新认领未通过）。 + YieldFailed(String), } impl std::fmt::Display for ServerError { @@ -56,6 +59,7 @@ impl std::fmt::Display for ServerError { ServerError::IdentityReadFailed(msg) => write!(f, "读取身份失败：{msg}"), ServerError::WorkerPanic(msg) => write!(f, "worker 线程异常退出：{msg}"), ServerError::SignalInstallFailed(msg) => write!(f, "信号处理器注册失败：{msg}"), + ServerError::YieldFailed(msg) => write!(f, "交接接管失败：{msg}"), } } } @@ -110,10 +114,12 @@ struct ServerContext { /// server 进程主体：认领 sidecar → 注册信号处理器 → 绑定 socket → 建内核 → /// 派生 worker → accept 循环 → 在途归零 → 清空发送端 → join worker → -/// drop 内核 → 删 sidecar 与 socket 文件。 +/// drop 内核 → 先删 socket 文件后删 sidecar。 /// /// # 步骤 -/// 1. 认领 sidecar（启动顺序契约：先写 sidecar 再 bind socket）。 +/// 1. 认领 sidecar（启动顺序契约：先写 sidecar 再 bind socket）；已有存活 +/// server 时转入交接路径（向旧 server 发 yield 请求，等旧 server 排空 +/// 退出后重新认领）。 /// 2. 注册信号处理器（绑定 socket 之前；失败回滚 sidecar，此时无存活线程）。 /// 3. 绑定监听器；失败且路径存在时删 stale socket 文件重试一次。 /// 4. 建内核（标准配置）与共享聚合计数器。 @@ -121,7 +127,8 @@ struct ServerContext { /// 6. 进入 accept 循环；关闭标志置位后按关闭序列退出。 /// /// # 边界与前提 -/// - 认领被拒（AlreadyRunning / RefusedAmbiguous）立即返回错误，不双注册。 +/// - 认领得 `AlreadyRunning` 时走交接路径，交接失败以 `YieldFailed` 返回。 +/// - 认领得 `RefusedAmbiguous` 立即返回错误，不双注册。 /// - 信号处理器注册失败（系统级错误）回滚 sidecar 并以错误返回。 /// - stale socket 文件删除重试后再失败时回滚删除 sidecar 并以错误返回。 /// @@ -132,14 +139,16 @@ struct ServerContext { /// /// # 注意事项 /// - 在调用线程上运行（后续 CLI 骨架任务在主线程分派到它）。 -/// - 关闭标志由关闭请求或 SIGINT/SIGTERM 信号置位。 +/// - 关闭标志由关闭请求、SIGINT/SIGTERM 信号或 yield 请求置位（三个触发源 +/// 汇合到同一条退出路径）。 /// - 关闭序列固定：清空发送端 → join 全部 worker → drop 内核（内核 drop /// 的调试断言要求存活 VM 数为零，worker 池随线程退出先释放，前提满足）。 pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { - // 认领 sidecar：已有存活 server 或存活证据不足即返回错误。 + // 认领 sidecar：已有存活 server 走交接路径，存活证据不足即返回错误。 match sidecar::claim_sidecar(&config.sidecar_path, &config.socket_path) { ClaimResult::Registered => {} - other => return Err(ServerError::ClaimRefused(other)), + ClaimResult::AlreadyRunning => yield_and_takeover(config)?, + ClaimResult::RefusedAmbiguous => return Err(ServerError::ClaimRefused(ClaimResult::RefusedAmbiguous)), } // 关闭标志提前创建并注册信号处理器：在绑定 socket 之前，socket 文件 @@ -196,13 +205,78 @@ pub fn run_server(config: &ServerConfig) -> Result<(), ServerError> { drop(ctx); drop(kernel); - // 删 sidecar 与 socket 文件（尽力而为，文件已不存在不视为错误）。 - let _ = fs::remove_file(&config.sidecar_path); + // 先删 socket 文件、后删 sidecar（尽力而为，文件已不存在不视为错误）： + // 新 server 以 sidecar 消失为接管信号，届时 socket 文件必已删除， + // 接管绑定无竞态。 let _ = fs::remove_file(&config.socket_path); + let _ = fs::remove_file(&config.sidecar_path); accept_result } +/// 交接接管：连接旧 server、发 yield 请求、等旧 server 排空退出后重新认领。 +/// +/// # 步骤 +/// 1. 连接旧 server 的 socket；连接失败判 `YieldFailed`。 +/// 2. 设 5 秒读超时，写 yield 请求帧。 +/// 3. 读一帧应答：读超时（旧 server 挂起）或应答非 yield 确认（旧版本 +/// binary 无该变体时以错误帧答复未知类型）判 `YieldFailed`。 +/// 4. 关闭连接，轮询 sidecar 文件消失（50 毫秒间隔、10 秒上限）；超时判 +/// `YieldFailed`。 +/// 5. 重新认领 sidecar；得 `Registered` 即接管成功，其余态判 `YieldFailed`。 +/// +/// # 边界与前提 +/// - 交接等待 10 秒长于旧 server 的排水安全超时 5 秒，正常路径必然先于 +/// 等待上限完成。 +/// - 旧 server 先删 socket 文件后删 sidecar，sidecar 消失时 socket 文件必 +/// 已删除，接管绑定无竞态。 +/// +/// # 副作用 +/// - 失败时不触碰旧 server 的文件（socket 与 sidecar），人工兜底归清理命令。 +/// +/// # 注意事项 +/// - 纯函数式入口：版本管理路径在版本不匹配强制交接时调用同一入口。 +fn yield_and_takeover(config: &ServerConfig) -> Result<(), ServerError> { + // 连接旧 server：连接成功是交接的前提。 + let mut stream = UnixStream::connect(&config.socket_path) + .map_err(|e| ServerError::YieldFailed(format!("连接旧 server 失败：{e}")))?; + + // 5 秒读超时：旧 server 挂起（不回应 yield）不无限阻塞。 + let _ = stream.set_read_timeout(Some(Duration::from_secs(5))); + stream + .write_all(protocol::encode_request(&ServerRequest::Yield).as_bytes()) + .map_err(|e| ServerError::YieldFailed(format!("写 yield 请求帧失败：{e}")))?; + + // 读确认帧：只有 yield 确认表示旧 server 已受理交接。 + let mut reader = FrameReader::new(io::BufReader::new(stream)); + let frame = reader + .read_frame() + .map_err(|e| ServerError::YieldFailed(format!("读 yield 确认帧失败：{e}")))? + .ok_or_else(|| ServerError::YieldFailed("旧 server 未应答即关闭连接".into()))?; + let response = protocol::parse_response(&frame) + .map_err(|e| ServerError::YieldFailed(format!("解析 yield 确认帧失败：{e}")))?; + if !matches!(response, ServerResponse::Yield) { + return Err(ServerError::YieldFailed(format!("旧 server 未受理交接：{response:?}"))); + } + // 关闭连接：旧 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)); + } + + // 重新认领：sidecar 已消失，O_EXCL 创建应成功；其余态判交接失败。 + match sidecar::claim_sidecar(&config.sidecar_path, &config.socket_path) { + ClaimResult::Registered => Ok(()), + other => Err(ServerError::YieldFailed(format!("交接后重新认领未通过：{other:?}"))), + } +} + /// 信号处理器体：置位关闭标志。 /// /// # 边界与前提 @@ -330,7 +404,7 @@ fn accept_loop( /// 2. 循环读帧：EOF、读超时、非法 UTF-8 退出；帧长超限按协议约定关连接。 /// 3. 畸形帧以 `Error` 帧答复后继续；执行请求路由到 worker 并阻塞等回复； /// 控制请求直接应答并写回。 -/// 4. 关闭确认帧写回即退出循环。 +/// 4. 关闭或 yield 确认帧写回即退出循环。 /// /// # 边界与前提 /// - 半开连接（对端静默不关）由 30 秒读超时有界化。 @@ -366,8 +440,8 @@ fn handle_connection(stream: UnixStream, ctx: &ServerContext, router: &WorkerRou break; } - // 关闭确认帧写回即退出循环。 - if matches!(response, ServerResponse::Shutdown) { + // 关闭或 yield 确认帧写回即退出循环。 + if matches!(response, ServerResponse::Shutdown | ServerResponse::Yield) { break; } } @@ -392,7 +466,7 @@ fn dispatch_eval(code: &str, max_steps: Option, router: &WorkerRouter) -> S } } -/// 控制请求分派：五类控制请求直接应答，不占 worker。 +/// 控制请求分派：六类控制请求直接应答，不占 worker。 /// /// Status 响应读全体 worker 池的共享聚合计数器。 fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerResponse { @@ -417,6 +491,11 @@ fn dispatch_control(request: &ServerRequest, ctx: &ServerContext) -> ServerRespo ctx.shutdown.store(true, Ordering::SeqCst); ServerResponse::Shutdown } + ServerRequest::Yield => { + // 置与关闭请求同一关闭标志，退出序列与关闭请求完全一致。 + ctx.shutdown.store(true, Ordering::SeqCst); + ServerResponse::Yield + } // 防御性兜底：正常路径 eval 在调用本函数前已被拦截走 worker 路由； // 若到达此臂说明上游分流缺失，以错误帧答复而非静默执行。 ServerRequest::Eval { .. } => ServerResponse::eval_err("执行请求应经 worker 路由，不应到达控制分派"), @@ -639,9 +718,10 @@ mod tests { stop_server(&mut stream, handle); } - /// 同一连接连续两帧（状态后版本）逐帧应答；同路径第二实例返回认领被拒错误。 + /// 同一连接连续两帧（状态后版本）逐帧应答；同路径第二实例走交接路径接管 + /// （第一 server 正常退出，第二实例接管 socket 路径）。 #[test] - fn multi_frame_same_connection_and_second_instance_refused() { + fn multi_frame_same_connection_and_second_instance_takeover() { let config = unique_paths("multi"); let handle = start_server(config.clone()); let mut stream = connect(&config); @@ -652,14 +732,33 @@ mod tests { let second = send_and_recv(&mut stream, &ServerRequest::Version); assert!(matches!(second, ServerResponse::Version { .. }), "应得 Version 帧：{second:?}"); - // 同路径第二实例：sidecar 已存在且 socket 存活，认领被拒。 - let refused = run_server(&config); - assert!( - matches!(refused, Err(ServerError::ClaimRefused(ClaimResult::AlreadyRunning))), - "第二实例应被拒：{refused:?}" - ); + // 关闭测试连接，让第一 server 的在途排水快速归零。 + drop(stream); - stop_server(&mut stream, handle); + // 同路径第二实例：sidecar 已存在且 socket 存活，走交接路径接管。 + let config_b = config.clone(); + let handle_b = std::thread::spawn(move || run_server(&config_b)); + + // 第一 server 应在有界时间内正常退出（yield 受理后走既有退出序列）。 + let deadline = Instant::now() + Duration::from_secs(30); + while !handle.is_finished() { + assert!(Instant::now() < deadline, "第一 server 应在 30 秒内退出"); + std::thread::sleep(Duration::from_millis(10)); + } + let result = handle.join().expect("第一 server 线程应正常退出"); + assert!(result.is_ok(), "第一 server 交接后应正常退出：{result:?}"); + + // 第二实例已接管：等 socket 文件出现（第二实例的绑定），核对 sidecar 身份。 + let deadline = Instant::now() + Duration::from_secs(30); + while !config.socket_path.exists() { + assert!(Instant::now() < deadline, "第二实例的 socket 文件应出现"); + std::thread::sleep(Duration::from_millis(10)); + } + let id = sidecar::read_identity(&config.sidecar_path).expect("第二实例的 sidecar 应可读"); + assert_eq!(id.pid, std::process::id(), "第二实例的 sidecar 应为本进程身份"); + + let mut stream_b = connect(&config); + stop_server(&mut stream_b, handle_b); } /// 端到端并发：两条连接各发 3 个 eval，6 个响应帧全部正确且逐连接顺序与请求一致。 @@ -789,4 +888,108 @@ mod tests { assert!(!config.sidecar_path.exists(), "sidecar 文件应被删除"); assert!(!config.socket_path.exists(), "socket 文件应被删除"); } + + /// 交接端到端：server A 运行中，server B 以同配置启动经 yield 路径接管； + /// A 正常退出，B 的 sidecar 为本进程身份（O_EXCL 重新认领通过即证明 A 的 + /// 文件已删除），B 可正常应答版本请求，随后以 Shutdown 停 B。 + #[test] + fn yield_takeover_end_to_end() { + let config = unique_paths("yield_e2e"); + let handle_a = start_server(config.clone()); + + // B 以同配置启动：认领得 AlreadyRunning 转交接路径。 + let config_b = config.clone(); + let handle_b = std::thread::spawn(move || run_server(&config_b)); + + // A 应在 30 秒内有界退出（排水加退出序列）。 + let deadline = Instant::now() + Duration::from_secs(30); + while !handle_a.is_finished() { + assert!(Instant::now() < deadline, "A 应在 30 秒内退出"); + std::thread::sleep(Duration::from_millis(10)); + } + let result_a = handle_a.join().expect("server A 线程应正常退出"); + assert!(result_a.is_ok(), "server A 交接后应正常退出：{result_a:?}"); + + // B 已接管：等 socket 文件出现（B 的绑定），再核对 sidecar 身份。 + let deadline = Instant::now() + Duration::from_secs(30); + while !config.socket_path.exists() { + assert!(Instant::now() < deadline, "B 的 socket 文件应出现"); + std::thread::sleep(Duration::from_millis(10)); + } + let id = sidecar::read_identity(&config.sidecar_path).expect("B 的 sidecar 应可读"); + assert_eq!(id.pid, std::process::id(), "B 的 sidecar 应为本进程身份"); + + // B 正常应答版本请求。 + let mut stream = connect(&config); + let response = send_and_recv(&mut stream, &ServerRequest::Version); + assert!( + matches!(response, ServerResponse::Version { .. }), + "B 应应答版本请求：{response:?}" + ); + stop_server(&mut stream, handle_b); + } + + /// 旧 server 不受理交接：裸监听者加手写 sidecar 模拟旧 server， + /// 接受连接后回 Error 帧（模拟旧版本 binary 无 yield 变体）； + /// run_server 判 YieldFailed 且不触碰旧 server 的文件。 + #[test] + fn yield_failed_when_old_server_refuses() { + let config = unique_paths("yield_refused"); + + // 模拟旧 server：绑定 socket 并写本进程身份 sidecar（PID 存活且启动 + // 时刻刻度匹配，认领判 AlreadyRunning）。 + let listener = UnixListener::bind(&config.socket_path).expect("绑定模拟旧 server 应成功"); + sidecar::write_exclusive( + &sidecar::ServerIdentity::new(config.socket_path.to_str().unwrap()), + &config.sidecar_path, + ) + .expect("写模拟 sidecar 应成功"); + + // 模拟连接处理：读一帧后回 Error 帧。 + 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::Error { + message: "未知协议类型：yield".into(), + }); + let _ = stream.write_all(frame.as_bytes()); + } + }); + + let err = run_server(&config).unwrap_err(); + assert!(matches!(err, ServerError::YieldFailed(_)), "应判 YieldFailed：{err:?}"); + assert!(config.sidecar_path.exists(), "模拟 sidecar 文件不应被删除"); + assert!(config.socket_path.exists(), "模拟 socket 文件不应被删除"); + } + + /// 旧 server 挂起：模拟旧 server 接受连接后不回应；run_server 在约 5 秒 + /// （读超时）后判 YieldFailed，且不触碰旧 server 的文件。 + #[test] + fn yield_failed_when_old_server_hangs() { + let config = unique_paths("yield_hang"); + + let listener = UnixListener::bind(&config.socket_path).expect("绑定模拟旧 server 应成功"); + sidecar::write_exclusive( + &sidecar::ServerIdentity::new(config.socket_path.to_str().unwrap()), + &config.sidecar_path, + ) + .expect("写模拟 sidecar 应成功"); + + // 模拟连接处理：接受连接后挂住不回应（模拟挂起的旧 server）。 + std::thread::spawn(move || { + if let Ok((stream, _)) = listener.accept() { + std::thread::sleep(Duration::from_secs(10)); + drop(stream); + } + }); + + let started = Instant::now(); + let err = run_server(&config).unwrap_err(); + let elapsed = started.elapsed(); + assert!(matches!(err, ServerError::YieldFailed(_)), "应判 YieldFailed：{err:?}"); + assert!(elapsed >= Duration::from_secs(5), "应命中 5 秒读超时：{elapsed:?}"); + assert!(config.sidecar_path.exists(), "模拟 sidecar 文件不应被删除"); + assert!(config.socket_path.exists(), "模拟 socket 文件不应被删除"); + } }