回扫的 terminated 改成按真实换行结尾判定
- DirectProjectHistoryReverseLines 增记 saw_delimiter，第一段按"文件是否以换行结尾"判定，之后的行按已切掉的分隔符判定 - 尾部残行（上一行完整、这一行没有换行）解析失败时改为跳过继续回扫，不再报错、也不再就地结束丢掉更早的完整历史 - 换行结尾的坏行（含文件第一段）与顺序读取保持一致地失败关闭 - 补两条回扫用例：尾部残行不丢更早条目、坏掉的第一段必须失败关闭（都按旧实现变异验证过）
This commit is contained in:
@@ -105,8 +105,9 @@ const DIRECT_PROJECT_HISTORY_REVERSE_SCAN_CHUNK_BYTES: usize = 16 * 1024;
|
||||
|
||||
/// 从文件尾向前回读到的一行。
|
||||
///
|
||||
/// `terminated` 表示这行后面确实有换行符：文件里只有末行可能没有换行，那种行是 append
|
||||
/// 侧承诺会修复的截断尾，解析失败时按"历史到此结束"处理，而不是报错。
|
||||
/// `terminated` 表示这行后面确实有换行符：整份历史里只有末尾那一段可能没有换行，那一段是
|
||||
/// append 侧承诺会修复的截断尾。解析失败时跳过这一行继续往前回扫，而不是报错——它是回扫
|
||||
/// 见到的第一行，一旦就地结束就会把后面完整的更早历史全部丢掉。
|
||||
struct DirectProjectHistoryReverseLine {
|
||||
bytes: Vec<u8>,
|
||||
terminated: bool,
|
||||
@@ -124,6 +125,8 @@ struct DirectProjectHistoryReverseLines {
|
||||
chunk: Vec<u8>,
|
||||
/// 下一次回读的起始偏移；0 表示文件头已经读完。
|
||||
position: u64,
|
||||
/// 是否已经切过一次换行：切过之后，接下来产出的行前面一定还有换行分隔。
|
||||
saw_delimiter: bool,
|
||||
}
|
||||
|
||||
impl DirectProjectHistoryReverseLines {
|
||||
@@ -145,6 +148,7 @@ impl DirectProjectHistoryReverseLines {
|
||||
pending: Vec::new(),
|
||||
chunk: vec![0u8; DIRECT_PROJECT_HISTORY_REVERSE_SCAN_CHUNK_BYTES],
|
||||
position,
|
||||
saw_delimiter: false,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -152,19 +156,27 @@ impl DirectProjectHistoryReverseLines {
|
||||
loop {
|
||||
if let Some(newline) = self.pending.iter().rposition(|byte| *byte == b'\n') {
|
||||
let bytes = self.pending[newline + 1..].to_vec();
|
||||
// `pending` 的尾部始终是文件尾，所以第一次切出来的这一段后面有没有换行，
|
||||
// 就是"整个文件是否以换行结尾"；之后切出来的行都被刚切掉的那个换行分隔。
|
||||
let terminated = if self.saw_delimiter {
|
||||
true
|
||||
} else {
|
||||
self.pending.last() == Some(&b'\n')
|
||||
};
|
||||
self.pending.truncate(newline);
|
||||
return Ok(Some(DirectProjectHistoryReverseLine {
|
||||
bytes,
|
||||
terminated: true,
|
||||
}));
|
||||
self.saw_delimiter = true;
|
||||
return Ok(Some(DirectProjectHistoryReverseLine { bytes, terminated }));
|
||||
}
|
||||
if self.position == 0 {
|
||||
if self.pending.is_empty() {
|
||||
return Ok(None);
|
||||
}
|
||||
// 走到这里说明剩下的字节里没有换行：只有文件根本没被换行分隔过，
|
||||
// 或者这就是第一行。前者是唯一没有换行结尾的行。
|
||||
let terminated = self.saw_delimiter;
|
||||
return Ok(Some(DirectProjectHistoryReverseLine {
|
||||
bytes: std::mem::take(&mut self.pending),
|
||||
terminated: false,
|
||||
terminated,
|
||||
}));
|
||||
}
|
||||
let read_len = usize::try_from(self.position)
|
||||
@@ -208,8 +220,9 @@ fn find_direct_project_history_item_by_id_at(
|
||||
}
|
||||
}
|
||||
Ok(None) => {}
|
||||
// 末行没有换行符：append 侧会修复这条截断尾，幂等回扫不能因此失败。
|
||||
Err(_) if !line.terminated => break,
|
||||
// 末行没有换行符：这是 append 侧会修复的截断尾，跳过它继续往前找，
|
||||
// 不能因为尾部半行就让整次幂等回扫失败。
|
||||
Err(_) if !line.terminated => continue,
|
||||
Err(error) => return Err(error),
|
||||
}
|
||||
}
|
||||
@@ -607,8 +620,9 @@ pub(crate) fn read_direct_project_history_items_slice_at(
|
||||
}
|
||||
let parsed: Value = match serde_json::from_slice(&line.bytes) {
|
||||
Ok(value) => value,
|
||||
// 与顺序读取一致：只有末行没有换行符时，解析失败才按"可修复的截断尾"结束。
|
||||
Err(_) if !line.terminated => break,
|
||||
// 与顺序读取一致：只有末行没有换行符时才是"可修复的截断尾"。回扫是从文件尾向前的，
|
||||
// 这一行是最新的一条，跳过它继续读更早的完整行，不能因此把整屏历史判成空的。
|
||||
Err(_) if !line.terminated => continue,
|
||||
Err(error) => {
|
||||
return Err(format!(
|
||||
"解析 DirectProject 历史失败：{}: {error}",
|
||||
@@ -838,6 +852,54 @@ mod tests {
|
||||
assert_eq!(first_item_id.as_deref(), Some("codex-item-1"));
|
||||
}
|
||||
|
||||
/// 判据：尾部残行（上一行完整、这一行没有换行结尾）跳过继续回扫，不报错也不就地结束。
|
||||
///
|
||||
/// 变异验证：`terminated` 按分支硬编码（旧实现）会把这条残行当成完整行，"读一屏"直接
|
||||
/// 报解析失败；就地结束则返回空窗口，把更早的完整历史全部丢掉。
|
||||
#[test]
|
||||
fn truncated_history_tail_is_skipped_without_losing_earlier_items() {
|
||||
let root = init_history_project("history-truncated-tail");
|
||||
let path = history_path(root.path());
|
||||
std::fs::create_dir_all(path.parent().expect("history parent")).expect("history dir");
|
||||
let first_row = RESPONSE_ITEM_ROW.replace("codex-item-2", "codex-item-1");
|
||||
let partial_tail = r#"{"type":"response_item","payload":{"type":"mess"#;
|
||||
std::fs::write(
|
||||
&path,
|
||||
format!("{first_row}\n{RESPONSE_ITEM_ROW}\n{partial_tail}"),
|
||||
)
|
||||
.expect("write truncated history tail");
|
||||
|
||||
let (items, has_more, _, first_item_id) =
|
||||
super::read_direct_project_history_items_slice_at(root.path(), None, 20).unwrap();
|
||||
let ids = items
|
||||
.iter()
|
||||
.filter_map(|item| item.get("id").and_then(Value::as_str))
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(ids, vec!["codex-item-1", "codex-item-2"]);
|
||||
assert!(!has_more);
|
||||
assert_eq!(first_item_id.as_deref(), Some("codex-item-1"));
|
||||
}
|
||||
|
||||
/// 判据：只有"末尾没有换行"那一段放宽；换行结尾的坏行一律失败关闭，第一段也不例外。
|
||||
///
|
||||
/// 变异验证：把第一段也当可修复尾行（旧实现）时，这条用例会静默返回空历史而不是报错。
|
||||
#[test]
|
||||
fn corrupt_first_line_fails_closed_like_any_newline_terminated_line() {
|
||||
let root = init_history_project("history-corrupt-head");
|
||||
let first_row = RESPONSE_ITEM_ROW.replace("codex-item-2", "codex-item-1");
|
||||
write_history_lines(
|
||||
root.path(),
|
||||
&[
|
||||
r#"{"type":"response_item","payload":{"type":"mess"#,
|
||||
&first_row,
|
||||
],
|
||||
);
|
||||
|
||||
let error = super::read_direct_project_history_items_slice_at(root.path(), None, 20)
|
||||
.expect_err("换行结尾的坏行必须失败关闭");
|
||||
assert!(error.starts_with("解析 DirectProject 历史失败"), "{error}");
|
||||
}
|
||||
|
||||
/// 判据：争用类失败会被"有界退避重试"真的吃掉，最终把条目落一行。
|
||||
///
|
||||
/// 注入标记是"让接下来 N 次单次尝试返回争用失败"；退避表只补一次重试，所以注入 1 次
|
||||
@@ -912,8 +974,8 @@ mod tests {
|
||||
/// 判据：格式类失败**不**触发重试。同一份历史每次读都是同一个结论，重试只是白等
|
||||
/// （而且会把"收尾失败"再拖一轮）。注入次数因此必须原样留着。
|
||||
///
|
||||
/// fixture 形状说明：回扫只对"后面还有换行的完整行"做严格解析，文件里第一段（以及末尾
|
||||
/// 没有换行的那一段）按可修复尾行放宽。所以损坏行必须夹在两条完整行中间才会失败关闭。
|
||||
/// fixture 形状说明：回扫只对"末尾没有换行的那一段"按可修复尾行放宽，其余每一行（包括
|
||||
/// 文件第一段）都必须有换行结尾并被严格解析。所以损坏行夹在两条完整行中间必然失败关闭。
|
||||
#[test]
|
||||
fn shape_failure_is_not_retried() {
|
||||
let root = init_history_project("shape-no-retry");
|
||||
|
||||
Reference in New Issue
Block a user