锁内只保留 append：DirectProject 历史的幂等回扫移出持锁窗口
- direct_project_history.rs：幂等回扫与序列化移到取锁之前，取锁后常态只做一次追加，不再在锁内读完并逐行解析整份历史 - direct_project_history.rs：回扫到取锁之间若历史文件变了（len/mtime 与回扫时不一致）才回到锁内重扫一次；同一 id 不写第二行的语义不放宽，只是常态不再为它读整份文件 - direct_project_history.rs：新增 direct_project_history_duplicate_at 三态判定（Absent/Identical/Conflict）与回扫状态快照，对外行为与原实现一致（幂等 no-op、id 冲突失败关闭） - direct_project_history.rs：新增回扫探针（两个时点回调）与三条用例：回扫在 append 锁外、锁外回扫后并发追加仍不重复写、id 冲突仍失败关闭 - project/agent_db.rs：新增测试专用探针 project_append_os_lock_is_held_for_test，用与生产同一套打开方式判断追加写目标的锁此刻是否被持有 - 变异验证：把回扫塞回锁内后 idempotency_reverse_scan_runs_outside_the_append_lock 变红（exit 101），报出「幂等回扫必须发生在 append 锁外」，验证后已还原
This commit is contained in:
@@ -4,10 +4,15 @@ use crate::project::{
|
||||
};
|
||||
use crate::{LocalConversationMessageRecord, LocalConversationResult};
|
||||
use serde_json::Value;
|
||||
use std::fs::File;
|
||||
use std::fs::{self, File};
|
||||
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
/// 回扫探针时点。生产编译后 `fire_direct_project_history_scan_probe` 是空实现，
|
||||
/// 只有测试会在这两个时点挂回调（见 `direct_project_history_scan_probe`）。
|
||||
const DIRECT_PROJECT_HISTORY_SCAN_PROBE_STARTED: usize = 0;
|
||||
const DIRECT_PROJECT_HISTORY_SCAN_PROBE_FINISHED: usize = 1;
|
||||
|
||||
const DIRECT_PROJECT_HISTORY_RECORD_TYPE: &str = "response_item";
|
||||
/// 格式切换到 `response_item`（#282）之前，DirectProject 主对话通过通用对话写入器
|
||||
/// 落到同一份 `project.jsonl`，行形状是 `PersistedLocalConversationMessageRecord`。
|
||||
@@ -96,6 +101,7 @@ fn find_direct_project_history_item_by_id_at(
|
||||
path: &Path,
|
||||
item_id: &str,
|
||||
) -> Result<Option<Value>, String> {
|
||||
fire_direct_project_history_scan_probe(DIRECT_PROJECT_HISTORY_SCAN_PROBE_STARTED);
|
||||
let mut file = File::open(path)
|
||||
.map_err(|error| format!("打开 DirectProject 历史失败：{}: {error}", path.display()))?;
|
||||
let mut position = file
|
||||
@@ -246,6 +252,86 @@ fn direct_project_legacy_row(parsed: &Value) -> Option<DirectProjectLegacyRow> {
|
||||
))
|
||||
}
|
||||
|
||||
/// 测试专用探针：在"锁外幂等回扫"这一段的两端回调。
|
||||
///
|
||||
/// 用来断言两件在别处无法观测的事实：① 回扫发生时 append 锁**不在**本进程手里；
|
||||
/// ② 回扫结束到真正 append 之间若发生并发追加，代码会回到锁内重扫。
|
||||
#[cfg(test)]
|
||||
mod direct_project_history_scan_probe {
|
||||
use std::sync::Mutex;
|
||||
|
||||
pub(super) const SCAN_STARTED: usize = super::DIRECT_PROJECT_HISTORY_SCAN_PROBE_STARTED;
|
||||
pub(super) const SCAN_FINISHED: usize = super::DIRECT_PROJECT_HISTORY_SCAN_PROBE_FINISHED;
|
||||
|
||||
static PROBES: Mutex<[Option<Box<dyn Fn() + Send>>; 2]> = Mutex::new([None, None]);
|
||||
|
||||
pub(super) fn install(slot: usize, probe: impl Fn() + Send + 'static) {
|
||||
PROBES.lock().expect("history scan probe lock")[slot] = Some(Box::new(probe));
|
||||
}
|
||||
|
||||
pub(super) fn fire(slot: usize) {
|
||||
let probe = PROBES.lock().expect("history scan probe lock")[slot].take();
|
||||
if let Some(probe) = probe {
|
||||
probe();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn fire_direct_project_history_scan_probe(slot: usize) {
|
||||
direct_project_history_scan_probe::fire(slot);
|
||||
}
|
||||
|
||||
#[cfg(not(test))]
|
||||
fn fire_direct_project_history_scan_probe(_slot: usize) {}
|
||||
|
||||
/// 回扫读到的历史文件状态。
|
||||
///
|
||||
/// 锁外回扫之后要复核"文件还是不是回扫时那一份"：只有一致才能沿用"这个 id 不在历史里"
|
||||
/// 的结论；不一致就回到锁内重扫一次（罕见）。于是锁内常态只剩 append。
|
||||
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
|
||||
struct DirectProjectHistoryScanState {
|
||||
len: u64,
|
||||
modified: Option<std::time::SystemTime>,
|
||||
}
|
||||
|
||||
fn direct_project_history_scan_state_at(
|
||||
path: &Path,
|
||||
) -> Result<DirectProjectHistoryScanState, String> {
|
||||
let metadata = fs::metadata(path).map_err(|error| {
|
||||
format!(
|
||||
"读取 DirectProject 历史元数据失败：{}: {error}",
|
||||
path.display()
|
||||
)
|
||||
})?;
|
||||
Ok(DirectProjectHistoryScanState {
|
||||
len: metadata.len(),
|
||||
modified: metadata.modified().ok(),
|
||||
})
|
||||
}
|
||||
|
||||
/// 幂等判定结果。`Absent` 表示历史里没有这个 id，可以继续追加。
|
||||
enum DirectProjectHistoryDuplicate {
|
||||
Absent,
|
||||
Identical,
|
||||
Conflict,
|
||||
}
|
||||
|
||||
fn direct_project_history_duplicate_at(
|
||||
path: &Path,
|
||||
item: &Value,
|
||||
item_id: &str,
|
||||
) -> Result<DirectProjectHistoryDuplicate, String> {
|
||||
if !prepare_game_creator_private_path_for_read(path, false, "DirectProject 历史")? {
|
||||
return Ok(DirectProjectHistoryDuplicate::Absent);
|
||||
}
|
||||
match find_direct_project_history_item_by_id_at(path, item_id)? {
|
||||
None => Ok(DirectProjectHistoryDuplicate::Absent),
|
||||
Some(existing) if &existing == item => Ok(DirectProjectHistoryDuplicate::Identical),
|
||||
Some(_) => Ok(DirectProjectHistoryDuplicate::Conflict),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn append_direct_project_history_item_at(
|
||||
root: &Path,
|
||||
item: &Value,
|
||||
@@ -275,23 +361,52 @@ fn append_direct_project_history_item_at_with_user_policy(
|
||||
if !allow_user_item && is_direct_project_codex_user_item(item) {
|
||||
return Ok(());
|
||||
}
|
||||
let _project_lock = crate::project::acquire_project_write_lock(root, "conversation.write")?;
|
||||
let path = history_path(root);
|
||||
let history_exists =
|
||||
prepare_game_creator_private_path_for_read(&path, false, "DirectProject 历史")?;
|
||||
let item_id = item.get("id").and_then(Value::as_str);
|
||||
// 幂等回扫与序列化都留在锁外。回扫（`find_direct_project_history_item_by_id_at`）
|
||||
// 会从文件尾逐行解析整份历史，新条目永远命不中，所以它天然是 O(文件大小) 的重活；
|
||||
// 历史到 MB 级时把它放进锁内，等于让另一个进程（`--agent-runner` 的 prompt 上下文、
|
||||
// MCP 宿主）等一次"读完整份历史"，而不是等一次追加。
|
||||
let scanned_state = if history_exists {
|
||||
let state = direct_project_history_scan_state_at(&path)?;
|
||||
if let Some(item_id) = item_id {
|
||||
match direct_project_history_duplicate_at(&path, item, item_id)? {
|
||||
DirectProjectHistoryDuplicate::Identical => return Ok(()),
|
||||
DirectProjectHistoryDuplicate::Conflict => {
|
||||
return Err(format!("DirectProject 历史 item id 冲突：{item_id}"))
|
||||
}
|
||||
DirectProjectHistoryDuplicate::Absent => {}
|
||||
}
|
||||
}
|
||||
Some(state)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
fire_direct_project_history_scan_probe(DIRECT_PROJECT_HISTORY_SCAN_PROBE_FINISHED);
|
||||
let line = record(item)?;
|
||||
let _project_lock = crate::project::acquire_project_write_lock(root, "conversation.write")?;
|
||||
let lock = project_append_lock_for(&path)?;
|
||||
let _append_guard = lock.lock("DirectProject 历史追加写")?;
|
||||
if history_exists {
|
||||
if let Some(item_id) = item.get("id").and_then(Value::as_str) {
|
||||
if let Some(existing) = find_direct_project_history_item_by_id_at(&path, item_id)? {
|
||||
if &existing == item {
|
||||
return Ok(());
|
||||
// 回扫到取锁之间可能有别的进程（或本项目另一条链）追加过：只有文件状态仍与回扫时一致
|
||||
// 才能沿用"这个 id 不在历史里"的结论，否则回到锁内重扫一次。
|
||||
// "同一个 id 不写第二行"的语义因此没有被放宽，只是常态不再为它读整份文件。
|
||||
if let Some(item_id) = item_id {
|
||||
let scan_is_stale = match scanned_state {
|
||||
Some(state) => direct_project_history_scan_state_at(&path)? != state,
|
||||
None => path.exists(),
|
||||
};
|
||||
if scan_is_stale {
|
||||
match direct_project_history_duplicate_at(&path, item, item_id)? {
|
||||
DirectProjectHistoryDuplicate::Identical => return Ok(()),
|
||||
DirectProjectHistoryDuplicate::Conflict => {
|
||||
return Err(format!("DirectProject 历史 item id 冲突：{item_id}"))
|
||||
}
|
||||
return Err(format!("DirectProject 历史 item id 冲突：{item_id}"));
|
||||
DirectProjectHistoryDuplicate::Absent => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
let line = record(item)?;
|
||||
append_jsonl_line_unlocked(&path, &line, "DirectProject 历史")
|
||||
}
|
||||
|
||||
@@ -429,11 +544,13 @@ pub(crate) fn read_direct_project_chat_history_at(
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
append_direct_project_history_item_at, append_direct_project_user_message_at, history_path,
|
||||
is_direct_project_internal_context_item, read_direct_project_chat_history_at,
|
||||
read_direct_project_history_items_at,
|
||||
append_direct_project_history_item_at, append_direct_project_user_message_at,
|
||||
direct_project_history_scan_probe, history_path, is_direct_project_internal_context_item,
|
||||
read_direct_project_chat_history_at, read_direct_project_history_items_at,
|
||||
};
|
||||
use serde_json::{json, Value};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
fn init_history_project(name: &str) -> tempfile::TempDir {
|
||||
let root = tempfile::tempdir().expect("temp project");
|
||||
@@ -452,6 +569,137 @@ mod tests {
|
||||
const LEGACY_ASSISTANT_ROW: &str = r#"{"schemaVersion":"game-creator-conversation.v1","role":"assistant","content":"已完成 第一行\n第二行 ","agentId":null,"updatedAt":1757000001}"#;
|
||||
const RESPONSE_ITEM_ROW: &str = r#"{"type":"response_item","payload":{"type":"message","role":"user","id":"codex-item-2","content":[{"type":"input_text","text":"再加一个按钮"}]}}"#;
|
||||
|
||||
/// 判据：幂等回扫发生时，append 锁**不能**在本进程手里。
|
||||
///
|
||||
/// 变异验证（把回扫挪回锁内）会把这条用例打成红的：探针在回扫时点用与生产同一套
|
||||
/// 打开方式（Windows 零共享 / Unix flock 非阻塞）去探同一把锁，持着就探不到。
|
||||
#[test]
|
||||
fn idempotency_reverse_scan_runs_outside_the_append_lock() {
|
||||
let root = init_history_project("scan-outside-lock");
|
||||
write_history_lines(
|
||||
root.path(),
|
||||
&[LEGACY_USER_ROW, RESPONSE_ITEM_ROW, LEGACY_ASSISTANT_ROW],
|
||||
);
|
||||
let path = history_path(root.path());
|
||||
let fired = Arc::new(AtomicBool::new(false));
|
||||
let lock_held = Arc::new(AtomicBool::new(false));
|
||||
let fired_probe = Arc::clone(&fired);
|
||||
let held_probe = Arc::clone(&lock_held);
|
||||
let probe_path = path.clone();
|
||||
direct_project_history_scan_probe::install(
|
||||
direct_project_history_scan_probe::SCAN_STARTED,
|
||||
move || {
|
||||
let held = crate::project::project_append_os_lock_is_held_for_test(&probe_path)
|
||||
.expect("probe append lock state");
|
||||
fired_probe.store(true, Ordering::SeqCst);
|
||||
held_probe.store(held, Ordering::SeqCst);
|
||||
},
|
||||
);
|
||||
|
||||
append_direct_project_history_item_at(
|
||||
root.path(),
|
||||
&json!({
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"id": "scan-window-item",
|
||||
"content": [{"type": "output_text", "text": "锁外回扫"}]
|
||||
}),
|
||||
)
|
||||
.expect("append item");
|
||||
|
||||
assert!(
|
||||
fired.load(Ordering::SeqCst),
|
||||
"幂等回扫必须真的发生，否则本用例只是空跑"
|
||||
);
|
||||
assert!(
|
||||
!lock_held.load(Ordering::SeqCst),
|
||||
"幂等回扫必须发生在 append 锁外：持锁读整份历史会把持锁窗口从一次追加拉成 O(文件大小)"
|
||||
);
|
||||
}
|
||||
|
||||
/// 判据：锁外回扫之后、真正 append 之前发生的并发追加，必须由"锁内重扫"兜住，
|
||||
/// 不能因为回扫结果过期就写下第二行同 id 记录。
|
||||
#[test]
|
||||
fn concurrent_append_after_the_out_of_lock_scan_still_prevents_a_duplicate_row() {
|
||||
let root = init_history_project("scan-stale-state");
|
||||
write_history_lines(root.path(), &[LEGACY_USER_ROW]);
|
||||
let path = history_path(root.path());
|
||||
let item = json!({
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"id": "stale-scan-item",
|
||||
"content": [{"type": "output_text", "text": "并发追加"}]
|
||||
});
|
||||
let injected_path = path.clone();
|
||||
let injected_item = item.clone();
|
||||
direct_project_history_scan_probe::install(
|
||||
direct_project_history_scan_probe::SCAN_FINISHED,
|
||||
move || {
|
||||
let line = serde_json::to_string(&json!({
|
||||
"type": "response_item",
|
||||
"payload": injected_item,
|
||||
}))
|
||||
.expect("serialize injected history row");
|
||||
let mut existing =
|
||||
std::fs::read_to_string(&injected_path).expect("read history before injection");
|
||||
existing.push_str(&line);
|
||||
existing.push('\n');
|
||||
std::fs::write(&injected_path, existing).expect("inject concurrent history row");
|
||||
},
|
||||
);
|
||||
|
||||
append_direct_project_history_item_at(root.path(), &item).expect("idempotent append");
|
||||
|
||||
let items = read_direct_project_history_items_at(root.path()).expect("read history");
|
||||
assert_eq!(
|
||||
items.iter().filter(|existing| *existing == &item).count(),
|
||||
1,
|
||||
"并发追加后必须由锁内重扫判定幂等，不能写下第二行"
|
||||
);
|
||||
}
|
||||
|
||||
/// 判据：同一个 id 但内容不同，在"回扫移到锁外"之后依然失败关闭且不追加。
|
||||
#[test]
|
||||
fn conflicting_item_id_still_fails_closed_with_the_scan_outside_the_lock() {
|
||||
let root = init_history_project("id-conflict-outside-lock");
|
||||
write_history_lines(root.path(), &[LEGACY_USER_ROW]);
|
||||
append_direct_project_history_item_at(
|
||||
root.path(),
|
||||
&json!({
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"id": "conflict-item",
|
||||
"content": [{"type": "output_text", "text": "第一次"}]
|
||||
}),
|
||||
)
|
||||
.expect("append first item");
|
||||
|
||||
let error = append_direct_project_history_item_at(
|
||||
root.path(),
|
||||
&json!({
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"id": "conflict-item",
|
||||
"content": [{"type": "output_text", "text": "同一个 id 但内容不同"}]
|
||||
}),
|
||||
)
|
||||
.expect_err("conflicting item id must fail closed");
|
||||
assert!(
|
||||
error.starts_with("DirectProject 历史 item id 冲突"),
|
||||
"{error}"
|
||||
);
|
||||
|
||||
let items = read_direct_project_history_items_at(root.path()).expect("read history");
|
||||
assert_eq!(
|
||||
items
|
||||
.iter()
|
||||
.filter(|item| item.get("id").and_then(Value::as_str) == Some("conflict-item"))
|
||||
.count(),
|
||||
1,
|
||||
"冲突条目不能追加第二行"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn filters_host_context_but_keeps_real_user_items() {
|
||||
assert!(is_direct_project_internal_context_item(&json!({
|
||||
|
||||
@@ -3623,6 +3623,26 @@ fn acquire_project_append_os_lock(path: &Path, error_label: &str) -> Result<File
|
||||
Err(format!("获取{error_label}跨进程锁超时：{}", path.display()))
|
||||
}
|
||||
|
||||
/// 测试专用探针：判断某个追加写目标此刻是否被别人持有 OS 锁。
|
||||
///
|
||||
/// 用与生产完全同一套打开方式（Windows 零共享句柄 / Unix `flock(LOCK_EX|LOCK_NB)`）去探同一
|
||||
/// 把锁文件：拿得到 ⇒ 没人持锁；拿不到 ⇒ 有人持锁。锁文件尚未创建同样算"没人持锁"（这里
|
||||
/// 刻意不创建它，探针不留副作用）。探针拿到的句柄立刻释放。
|
||||
#[cfg(test)]
|
||||
pub(crate) fn project_append_os_lock_is_held_for_test(path: &Path) -> Result<bool, String> {
|
||||
let lock_path = project_append_os_lock_path(path)?;
|
||||
if !lock_path.exists() {
|
||||
return Ok(false);
|
||||
}
|
||||
match try_open_project_append_os_lock(&lock_path, "追加写锁探针")? {
|
||||
Some(file) => {
|
||||
drop(file);
|
||||
Ok(false)
|
||||
}
|
||||
None => Ok(true),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn try_open_project_append_os_lock(path: &Path, error_label: &str) -> Result<Option<File>, String> {
|
||||
use std::os::fd::AsRawFd;
|
||||
|
||||
Reference in New Issue
Block a user