From eadd6541c4c70e5f971f61968b9d36c8da155869 Mon Sep 17 00:00:00 2001 From: pjh456 <147148383@qq.com> Date: Sun, 11 Oct 2026 17:17:38 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E5=88=86=E7=89=87?= =?UTF-8?q?=E5=B8=A7=E5=8D=8F=E8=AE=AE=E6=A8=A1=E5=9D=97=E4=B8=8E=E7=BC=96?= =?UTF-8?q?=E7=A0=81=E8=A7=A3=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 41 +- Cargo.toml | 2 +- crates/oxide_test262/Cargo.toml | 1 + crates/oxide_test262/src/main.rs | 1 + crates/oxide_test262/src/shard_protocol.rs | 491 +++++++++++++++++++++ 5 files changed, 515 insertions(+), 21 deletions(-) create mode 100644 crates/oxide_test262/src/shard_protocol.rs diff --git a/Cargo.lock b/Cargo.lock index 209d7636..42d3a731 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1160,7 +1160,7 @@ dependencies = [ [[package]] name = "oxide_builtins" -version = "0.15.16" +version = "0.15.17" dependencies = [ "chrono", "fast-float", @@ -1185,7 +1185,7 @@ dependencies = [ [[package]] name = "oxide_bytecode" -version = "0.15.16" +version = "0.15.17" dependencies = [ "num-bigint", "oxide_log", @@ -1193,7 +1193,7 @@ dependencies = [ [[package]] name = "oxide_cfg" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxide_bytecode", "oxide_emit", @@ -1204,7 +1204,7 @@ dependencies = [ [[package]] name = "oxide_cli" -version = "0.15.16" +version = "0.15.17" dependencies = [ "ansi_term", "bumpalo", @@ -1228,7 +1228,7 @@ dependencies = [ [[package]] name = "oxide_code_cache" -version = "0.15.16" +version = "0.15.17" dependencies = [ "lru", "oxide_bytecode", @@ -1237,7 +1237,7 @@ dependencies = [ [[package]] name = "oxide_compiler" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1257,7 +1257,7 @@ dependencies = [ [[package]] name = "oxide_dce" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1271,7 +1271,7 @@ dependencies = [ [[package]] name = "oxide_emit" -version = "0.15.16" +version = "0.15.17" dependencies = [ "num-bigint", "oxc_allocator", @@ -1286,7 +1286,7 @@ dependencies = [ [[package]] name = "oxide_engine" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1298,7 +1298,7 @@ dependencies = [ [[package]] name = "oxide_ir" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxide_bytecode", "oxide_log", @@ -1307,7 +1307,7 @@ dependencies = [ [[package]] name = "oxide_kernel" -version = "0.15.16" +version = "0.15.17" dependencies = [ "bumpalo", "dashmap", @@ -1321,7 +1321,7 @@ dependencies = [ [[package]] name = "oxide_liveness" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1333,7 +1333,7 @@ dependencies = [ [[package]] name = "oxide_log" -version = "0.15.16" +version = "0.15.17" dependencies = [ "tracing", "tracing-appender", @@ -1342,11 +1342,11 @@ dependencies = [ [[package]] name = "oxide_net" -version = "0.15.16" +version = "0.15.17" [[package]] name = "oxide_parser" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxc_allocator", "oxc_ast", @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "oxide_regalloc" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxide_bytecode", "oxide_cfg", @@ -1377,7 +1377,7 @@ dependencies = [ [[package]] name = "oxide_runtime_api" -version = "0.15.16" +version = "0.15.17" dependencies = [ "num-bigint", "num-traits", @@ -1390,7 +1390,7 @@ dependencies = [ [[package]] name = "oxide_test262" -version = "0.15.16" +version = "0.15.17" dependencies = [ "oxide_bytecode", "oxide_compiler", @@ -1402,13 +1402,14 @@ dependencies = [ "oxide_types", "oxide_vm", "serde", + "serde_json", "serde_yaml", "walkdir", ] [[package]] name = "oxide_types" -version = "0.15.16" +version = "0.15.17" dependencies = [ "bumpalo", "dashmap", @@ -1421,7 +1422,7 @@ dependencies = [ [[package]] name = "oxide_vm" -version = "0.15.16" +version = "0.15.17" dependencies = [ "bumpalo", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 96211a21..24c0899b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.15.16" +version = "0.15.17" edition = "2021" license = "MIT" rust-version = "1.80" diff --git a/crates/oxide_test262/Cargo.toml b/crates/oxide_test262/Cargo.toml index 872ac9de..8ceaafbb 100644 --- a/crates/oxide_test262/Cargo.toml +++ b/crates/oxide_test262/Cargo.toml @@ -21,6 +21,7 @@ oxide_engine = { path = "../oxide_engine" } oxide_types = { path = "../oxide_types" } oxide_log = { path = "../oxide_log" } serde = { version = "1", features = ["derive"] } +serde_json.workspace = true serde_yaml = "0.9" walkdir = "2" diff --git a/crates/oxide_test262/src/main.rs b/crates/oxide_test262/src/main.rs index 1d0af95e..d81aa02d 100644 --- a/crates/oxide_test262/src/main.rs +++ b/crates/oxide_test262/src/main.rs @@ -14,6 +14,7 @@ mod judge; mod meta; mod report; mod runner; +mod shard_protocol; mod stats; mod supervise; mod test262_log; diff --git a/crates/oxide_test262/src/shard_protocol.rs b/crates/oxide_test262/src/shard_protocol.rs new file mode 100644 index 00000000..f0584c20 --- /dev/null +++ b/crates/oxide_test262/src/shard_protocol.rs @@ -0,0 +1,491 @@ +//! 分片帧协议：远程分片的 NDJSON 帧格式与 serde 结构。 +//! +//! 帧约定：每帧一行 JSON，以换行符（U+000A）终止，UTF-8 编码。 +//! 两个方向的帧均由 `type` 字段区分类型（tagged 枚举，snake_case 取值）： +//! `ShardCommand` 是监督者到 worker 方向（assign / kill），`ShardEvent` 是 +//! worker 到监督者方向（heartbeat / fail / scene / done）。 +//! 帧长上限 `MAX_FRAME_BYTES`；超限帧读侧提前停止，避免无界内存增长。 +//! 畸形帧（非法 JSON、未知类型、载荷不匹配、空帧）统一返回 `ShardProtocolError`， +//! 任何路径不 panic。 +//! +//! 写序契约由 worker 侧执行：同一测试的心跳帧先于失败帧，协议模块本身无状态。 +//! assign 帧携带窗口路径清单切片：worker 读不到监督者本机临时文件，窗口内容 +//! 必须随帧携带。 +//! +//! 本模块是纯协议定义层，调用方（worker 模式与远程监督者）尚未接线， +//! 豁免死代码告警。 +#![allow(dead_code)] + +use std::collections::HashMap; +use std::io::BufRead; + +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; + +/// 帧长上限（字节，1 MiB）：超过即判 `FrameTooLarge`，调用方应关闭连接。 +pub(crate) const MAX_FRAME_BYTES: usize = 1024 * 1024; + +/// 协议错误：畸形帧或未知类型，统一返回、不 panic。 +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum ShardProtocolError { + /// 帧不是合法 JSON，附语法错误描述。 + InvalidJson(String), + /// 帧是合法 JSON，但 `type` 字段缺失或不是已知协议类型，附原始取值。 + UnknownType(String), + /// `type` 已知但载荷字段与结构不符（缺字段、类型不匹配），附错误描述。 + MalformedPayload(String), + /// 帧为空（仅空白字符）。 + EmptyFrame, + /// 帧超过长度上限（见 `MAX_FRAME_BYTES`）。 + FrameTooLarge, +} + +impl std::fmt::Display for ShardProtocolError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ShardProtocolError::InvalidJson(msg) => write!(f, "帧不是合法 JSON：{msg}"), + ShardProtocolError::UnknownType(tag) => write!(f, "未知协议类型：{tag}"), + ShardProtocolError::MalformedPayload(msg) => write!(f, "帧载荷不匹配：{msg}"), + ShardProtocolError::EmptyFrame => write!(f, "帧为空"), + ShardProtocolError::FrameTooLarge => write!(f, "帧超过长度上限"), + } + } +} + +/// 命令帧（监督者到 worker）：窗口指派与 kill。 +/// +/// 序列化为 tagged 对象，`type` 字段区分类型（snake_case）。 +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub(crate) enum ShardCommand { + /// 窗口指派：携带该窗口的路径清单切片与单测试超时。 + /// `paths` 是窗口路径清单切片（worker 读不到监督者本机临时文件，须随帧携带）； + /// `timeout_secs` 是单测试超时，供 worker 侧看门狗使用。 + Assign { + wstart: usize, + wend: usize, + paths: Vec, + timeout_secs: u64, + }, + /// kill：无载荷。 + Kill, +} + +/// 事件帧（worker 到监督者）：心跳、失败记录、现场、窗口结束。 +/// +/// 序列化为 tagged 对象，`type` 字段区分类型（snake_case）。 +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub(crate) enum ShardEvent { + /// 心跳：字段与心跳记录逐一对应；`categories` 是失败分类计数快照， + /// 序列化为 JSON 对象（键可含冒号与空格）。 + Heartbeat { + phase: String, + index: usize, + pass: usize, + fail: usize, + skip: usize, + categories: HashMap, + hb_write_errors: usize, + }, + /// 失败记录：一测试失败一帧，字段与旁路失败行四元组逐一对应。 + Fail { + index: usize, + category: String, + subkey: String, + message: String, + }, + /// 现场：last-pc 现场行，加可选的 sidecar 日志尾。 + Scene { + line: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + log_tail: Option, + }, + /// 窗口结束：无载荷（窗口结束下标由 assign 帧的 wend 隐含）。 + Done, +} + +/// 命令帧可取的 `type` 值（须与 `ShardCommand` 变体保持同步）。 +const COMMAND_TAGS: &[&str] = &["assign", "kill"]; + +/// 事件帧可取的 `type` 值（须与 `ShardEvent` 变体保持同步）。 +const EVENT_TAGS: &[&str] = &["heartbeat", "fail", "scene", "done"]; + +/// 序列化命令为帧字符串（含结尾换行）。 +/// +/// 命令结构字段皆为基础类型，序列化不会失败；`expect` 是安全网而非错误路径。 +pub(crate) fn encode_command(cmd: &ShardCommand) -> String { + let mut frame = serde_json::to_string(cmd).expect("命令结构序列化不会失败"); + frame.push('\n'); + frame +} + +/// 序列化事件为帧字符串（含结尾换行）。 +/// +/// 事件结构字段皆为基础类型，序列化不会失败；`expect` 是安全网而非错误路径。 +pub(crate) fn encode_event(ev: &ShardEvent) -> String { + let mut frame = serde_json::to_string(ev).expect("事件结构序列化不会失败"); + frame.push('\n'); + frame +} + +/// 解析单行命令帧（结尾换行可有可无）。 +/// +/// # 步骤 +/// 1. 去首尾空白；空帧判 `EmptyFrame`，超长判 `FrameTooLarge`。 +/// 2. 解析为 JSON；语法错误判 `InvalidJson`。 +/// 3. 读取 `type` 字段；缺失或未知判 `UnknownType`（附原始取值）。 +/// 4. 反序列化为 `ShardCommand`；载荷不匹配判 `MalformedPayload`。 +/// +/// # 边界与前提 +/// - 输入是单帧（一行）；分行由读侧负责（见 `FrameReader`）。 +/// - 未知类型保留原始 `type` 取值，供调用方引用。 +pub(crate) fn parse_command(line: &str) -> Result { + parse_frame(line, COMMAND_TAGS) +} + +/// 解析单行事件帧（结尾换行可有可无）。 +/// +/// 步骤与判定同 `parse_command`，目标结构为 `ShardEvent`。 +pub(crate) fn parse_event(line: &str) -> Result { + parse_frame(line, EVENT_TAGS) +} + +fn parse_frame(line: &str, known_tags: &[&str]) -> Result { + // 去首尾空白：帧自带的换行与行尾空白不参与解析。 + let trimmed = line.trim(); + if trimmed.is_empty() { + return Err(ShardProtocolError::EmptyFrame); + } + if trimmed.len() > MAX_FRAME_BYTES { + return Err(ShardProtocolError::FrameTooLarge); + } + + // 先解析为通用 Value，以便先检查 `type` 字段再定型。 + let value: serde_json::Value = + serde_json::from_str(trimmed).map_err(|e| ShardProtocolError::InvalidJson(e.to_string()))?; + + // 读取 `type` 字段：缺失或非字符串取值原样上报。 + let tag = match value.get("type") { + None => "".to_string(), + Some(t) => t.as_str().map(str::to_string).unwrap_or_else(|| t.to_string()), + }; + if !known_tags.contains(&tag.as_str()) { + return Err(ShardProtocolError::UnknownType(tag)); + } + + // 类型已知：反序列化为目标结构，载荷不匹配判 `MalformedPayload`。 + serde_json::from_value(value).map_err(|e| ShardProtocolError::MalformedPayload(e.to_string())) +} + +/// 帧读取器：从行流读取 NDJSON 帧，跳过空行。 +/// +/// 流可能一次交还超过一行的数据（带缓冲的读侧），读取器内部持有余量缓冲， +/// 调用方不得绕过读取器直接读底层流。 +pub(crate) struct FrameReader { + inner: R, + /// 上次读取后尚未消费成帧的余量数据。 + pending: Vec, +} + +impl FrameReader { + /// 以底层流构造读取器。 + pub(crate) fn new(inner: R) -> Self { + FrameReader { inner, pending: Vec::new() } + } + + /// 读取一帧（跳过空行）。 + /// + /// # 步骤 + /// 1. 取一行：优先消费余量缓冲，不足时按 4 KiB 块从流补足到换行或 EOF。 + /// 2. 行长超过上限即提前停止读取，流停在行中间，调用方应判 `FrameTooLarge` + /// 并关闭连接，不得在同一流上继续读。 + /// 3. 空行（仅空白）跳过；EOF 返回 `Ok(None)`。 + /// + /// # 边界与前提 + /// - 帧必须是合法 UTF-8，否则返回 `InvalidData` io 错误。 + /// - 超长提前停止后流停在行中间，调用方必须关闭连接。 + /// + /// # 副作用 + /// - 只从流读取，并维护余量缓冲。 + pub(crate) fn read_frame(&mut self) -> std::io::Result> { + loop { + let line = match self.take_line()? { + None => return Ok(None), + Some(line) => line, + }; + + // UTF-8 校验：协议帧必须是合法 UTF-8 文本。 + let text = String::from_utf8(line) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidData, "帧不是合法 UTF-8"))?; + if text.trim().is_empty() { + continue; + } + return Ok(Some(text)); + } + } + + /// 取一行（含结尾换行），或 EOF 时的残行。 + /// + /// 余量缓冲已含换行时直接切出首行；缓冲无换行且超上限时提前停止； + /// 否则从流读一块追加，直到凑齐一行或 EOF。 + fn take_line(&mut self) -> std::io::Result>> { + loop { + // 余量缓冲已含换行：切出首行，余下进下一轮。 + if let Some(pos) = self.pending.iter().position(|&b| b == b'\n') { + let line = std::mem::take(&mut self.pending); + let (first, rest) = line.split_at(pos + 1); + self.pending = rest.to_vec(); + return Ok(Some(first.to_vec())); + } + + // 余量缓冲无换行且已超上限：行必然超长，提前停止。 + if self.pending.len() > MAX_FRAME_BYTES { + return Ok(Some(std::mem::take(&mut self.pending))); + } + + // 从流读一块追加到余量缓冲。 + // 信号中断（EINTR）不视为错误：重试读，保持读循环不变式。 + let mut chunk = [0u8; 4096]; + let n = match self.inner.read(&mut chunk) { + Ok(n) => n, + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + Err(e) => return Err(e), + }; + if n == 0 { + // EOF：余量缓冲的残行原样交付（可能为空）。 + let line = std::mem::take(&mut self.pending); + return Ok(if line.is_empty() { None } else { Some(line) }); + } + self.pending.extend_from_slice(&chunk[..n]); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + + /// 命令帧 round-trip：编码后解析应得回原结构。 + fn roundtrip_command(cmd: &ShardCommand) { + let frame = encode_command(cmd); + assert!(frame.ends_with('\n'), "帧必须以换行终止"); + let parsed = parse_command(&frame).expect("命令帧 round-trip 应成功"); + assert_eq!(parsed, *cmd); + } + + /// 事件帧 round-trip：编码后解析应得回原结构。 + fn roundtrip_event(ev: &ShardEvent) { + let frame = encode_event(ev); + assert!(frame.ends_with('\n'), "帧必须以换行终止"); + let parsed = parse_event(&frame).expect("事件帧 round-trip 应成功"); + assert_eq!(parsed, *ev); + } + + /// 命令 round-trip：assign（带路径清单与超时）与 kill。 + #[test] + fn command_roundtrip() { + roundtrip_command(&ShardCommand::Assign { + wstart: 0, + wend: 10, + paths: vec!["a/b.js".into(), "c/d.js".into()], + timeout_secs: 10, + }); + roundtrip_command(&ShardCommand::Kill); + } + + /// 事件 round-trip：心跳（带失败类别）、失败记录、现场（带与不带日志尾）、窗口结束。 + #[test] + fn event_roundtrip() { + let mut categories = HashMap::new(); + categories.insert("vm: not callable".to_string(), 2); + roundtrip_event(&ShardEvent::Heartbeat { + phase: "COMPLETED".into(), + index: 3, + pass: 2, + fail: 1, + skip: 0, + categories, + hb_write_errors: 1, + }); + roundtrip_event(&ShardEvent::Fail { + index: 5, + category: "vm: not callable".into(), + subkey: "x".into(), + message: "x is not callable".into(), + }); + roundtrip_event(&ShardEvent::Scene { + line: "pc=1 op=ADD frames=1".into(), + log_tail: None, + }); + roundtrip_event(&ShardEvent::Scene { + line: "pc=1 op=ADD frames=1".into(), + log_tail: Some("line a\nline b".into()), + }); + roundtrip_event(&ShardEvent::Done); + } + + /// 序列化形态：钉住精确 JSON 字符串，含 categories 键带冒号与空格的形态。 + #[test] + fn serialization_shape() { + assert_eq!( + serde_json::to_string(&ShardCommand::Assign { + wstart: 0, + wend: 10, + paths: vec!["a/b.js".into(), "c/d.js".into()], + timeout_secs: 10, + }) + .unwrap(), + r#"{"type":"assign","wstart":0,"wend":10,"paths":["a/b.js","c/d.js"],"timeout_secs":10}"# + ); + assert_eq!(serde_json::to_string(&ShardCommand::Kill).unwrap(), r#"{"type":"kill"}"#); + + let mut categories = HashMap::new(); + categories.insert("vm: not callable".to_string(), 1); + assert_eq!( + serde_json::to_string(&ShardEvent::Heartbeat { + phase: "COMPLETED".into(), + index: 3, + pass: 2, + fail: 1, + skip: 0, + categories, + hb_write_errors: 0, + }) + .unwrap(), + r#"{"type":"heartbeat","phase":"COMPLETED","index":3,"pass":2,"fail":1,"skip":0,"categories":{"vm: not callable":1},"hb_write_errors":0}"# + ); + assert_eq!( + serde_json::to_string(&ShardEvent::Fail { + index: 5, + category: "vm: not callable".into(), + subkey: "x".into(), + message: "x is not callable".into(), + }) + .unwrap(), + r#"{"type":"fail","index":5,"category":"vm: not callable","subkey":"x","message":"x is not callable"}"# + ); + assert_eq!( + serde_json::to_string(&ShardEvent::Scene { + line: "pc=1 op=ADD frames=1".into(), + log_tail: None + }) + .unwrap(), + r#"{"type":"scene","line":"pc=1 op=ADD frames=1"}"# + ); + assert_eq!( + serde_json::to_string(&ShardEvent::Scene { + line: "pc=1 op=ADD frames=1".into(), + log_tail: Some("a\nb".into()), + }) + .unwrap(), + r#"{"type":"scene","line":"pc=1 op=ADD frames=1","log_tail":"a\nb"}"# + ); + assert_eq!(serde_json::to_string(&ShardEvent::Done).unwrap(), r#"{"type":"done"}"#); + } + + /// 非法 JSON：判 `InvalidJson`，不 panic。 + #[test] + fn malformed_json() { + let err = parse_command("not json at all").unwrap_err(); + assert!(matches!(err, ShardProtocolError::InvalidJson(_)), "应判 InvalidJson：{err}"); + let err = parse_event("not json at all").unwrap_err(); + assert!(matches!(err, ShardProtocolError::InvalidJson(_)), "应判 InvalidJson：{err}"); + } + + /// 未知 `type`：判 `UnknownType` 且保留原始取值。 + #[test] + fn unknown_type() { + let err = parse_command(r#"{"type":"bogus"}"#).unwrap_err(); + assert_eq!(err, ShardProtocolError::UnknownType("bogus".into())); + let err = parse_event(r#"{"type":"bogus"}"#).unwrap_err(); + assert_eq!(err, ShardProtocolError::UnknownType("bogus".into())); + } + + /// 缺失 `type`：判 `UnknownType` 且取值为 ``。 + #[test] + fn missing_type() { + let err = parse_command(r#"{"wstart":0}"#).unwrap_err(); + assert_eq!(err, ShardProtocolError::UnknownType("".into())); + let err = parse_event(r#"{"index":1}"#).unwrap_err(); + assert_eq!(err, ShardProtocolError::UnknownType("".into())); + } + + /// 载荷不匹配：`type` 已知但字段缺失或类型错，判 `MalformedPayload`。 + #[test] + fn malformed_payload() { + let err = parse_command(r#"{"type":"assign"}"#).unwrap_err(); + assert!(matches!(err, ShardProtocolError::MalformedPayload(_)), "应判 MalformedPayload：{err}"); + let err = parse_command(r#"{"type":"assign","wstart":0,"wend":10,"paths":[],"timeout_secs":"x"}"#).unwrap_err(); + assert!(matches!(err, ShardProtocolError::MalformedPayload(_)), "应判 MalformedPayload：{err}"); + let err = parse_event(r#"{"type":"fail"}"#).unwrap_err(); + assert!(matches!(err, ShardProtocolError::MalformedPayload(_)), "应判 MalformedPayload：{err}"); + let err = parse_event( + r#"{"type":"heartbeat","phase":"COMPLETED","index":1,"pass":0,"fail":0,"skip":0,"categories":{}}"#, + ) + .unwrap_err(); + assert!(matches!(err, ShardProtocolError::MalformedPayload(_)), "应判 MalformedPayload：{err}"); + } + + /// 空帧（空串与纯空白）：判 `EmptyFrame`。 + #[test] + fn empty_frame() { + assert_eq!(parse_command("").unwrap_err(), ShardProtocolError::EmptyFrame); + assert_eq!(parse_command(" \t ").unwrap_err(), ShardProtocolError::EmptyFrame); + assert_eq!(parse_event("").unwrap_err(), ShardProtocolError::EmptyFrame); + } + + /// 超长帧：判 `FrameTooLarge`。 + #[test] + fn oversized_frame() { + let line = format!("{{\"type\":\"kill\",\"pad\":\"{}\"}}", "a".repeat(MAX_FRAME_BYTES)); + assert_eq!(parse_command(&line).unwrap_err(), ShardProtocolError::FrameTooLarge); + let line = format!("{{\"type\":\"done\",\"pad\":\"{}\"}}", "a".repeat(MAX_FRAME_BYTES)); + assert_eq!(parse_event(&line).unwrap_err(), ShardProtocolError::FrameTooLarge); + } + + /// 帧读取：跳过空行，返回非空帧（含结尾换行）；流一次交还多行时不丢数据。 + #[test] + fn read_frame_skips_blank_lines() { + let mut reader = FrameReader::new(Cursor::new(b"\n \n{\"type\":\"kill\"}\n".to_vec())); + let frame = reader.read_frame().expect("读帧应成功").expect("应读到帧"); + assert_eq!(frame, "{\"type\":\"kill\"}\n"); + assert_eq!(reader.read_frame().expect("EOF 应得 None"), None); + } + + /// 帧读取：一次读取交还多帧时逐帧交付，余量不丢。 + #[test] + fn read_frame_multiple_frames_in_one_read() { + let mut reader = FrameReader::new(Cursor::new(b"{\"type\":\"kill\"}\n{\"type\":\"done\"}\n".to_vec())); + assert_eq!(reader.read_frame().expect("读帧应成功").expect("应读到帧"), "{\"type\":\"kill\"}\n"); + assert_eq!(reader.read_frame().expect("读帧应成功").expect("应读到帧"), "{\"type\":\"done\"}\n"); + assert_eq!(reader.read_frame().expect("EOF 应得 None"), None); + } + + /// 帧读取：EOF 前无换行的残行也作为一帧交付。 + #[test] + fn read_frame_unterminated_line_at_eof() { + let mut reader = FrameReader::new(Cursor::new(b"{\"type\":\"done\"}".to_vec())); + let frame = reader.read_frame().expect("读帧应成功").expect("应读到帧"); + assert_eq!(frame, "{\"type\":\"done\"}"); + } + + /// 帧读取：超长行提前停止，返回长度超上限的残行，解析判 `FrameTooLarge`。 + #[test] + fn read_frame_oversized_stops_early() { + let payload = vec![b'a'; MAX_FRAME_BYTES + 8192]; + let mut reader = FrameReader::new(Cursor::new(payload)); + let frame = reader.read_frame().expect("读帧应成功").expect("应读到帧"); + assert!(frame.len() > MAX_FRAME_BYTES, "超长行应提前停止并返回残行"); + assert_eq!(parse_command(&frame).unwrap_err(), ShardProtocolError::FrameTooLarge); + } + + /// 帧读取：非法 UTF-8 判 `InvalidData` io 错误，不 panic。 + #[test] + fn read_frame_non_utf8() { + let mut reader = FrameReader::new(Cursor::new(vec![0xff, 0xfe, b'\n'])); + let err = reader.read_frame().unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + } +}