优化: MCP 收尾(update原子CAS治TOCTOU/advance_task数据变更映射/list_trash分页) + miniapp渲染收尾(mention chip可视化/代码块语言标签) + 销账
This commit is contained in:
+98
-15
@@ -140,7 +140,7 @@ pub fn all_tools() -> &'static Vec<&'static ToolSpec> {
|
||||
"confidence": opt_str_field("置信度(可空:high/medium/low)")
|
||||
}), &["kind", "title", "content"]), Medium, insert_knowledge),
|
||||
// ─── 回收站 ───
|
||||
spec("list_trash", "列出回收站(deleted_at IS NOT NULL 的项目与任务)", object_schema(json!({}), &[]), Low, list_trash),
|
||||
spec("list_trash", "列出回收站(deleted_at IS NOT NULL 的项目与任务;分页 offset/limit,默认 limit=50 上限 100,projects/tasks 各自独立分页)", object_schema(json!({"offset": int_field("偏移量(可空,默认 0)"), "limit": int_field("返回上限(可空,默认 50,上限 100)")}), &[]), Low, list_trash),
|
||||
spec("restore_project", "从回收站恢复项目(Medium 风险+审计日志)", object_schema(json!({"id": str_field("项目 ID")}), &["id"]), Medium, restore_project),
|
||||
]
|
||||
})
|
||||
@@ -440,12 +440,15 @@ fn update_project(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult>
|
||||
created_at: existing.created_at,
|
||||
updated_at: now,
|
||||
};
|
||||
match repo.update_full(&rec).await {
|
||||
// 权威 CAS:以 get_by_id 刚读到的 updated_at 为 expected,单条原子条件写。
|
||||
// 与 check_expected_updated_at(客户端显式 expected_updated_at 的提前拦截)不同,
|
||||
// 这里封死「读 existing → 写 rec」之间的跨进程竞态窗口:并发端(GUI)已改则 affected==0。
|
||||
match repo.update_full_cas(&rec, &existing.updated_at).await {
|
||||
Ok(true) => {
|
||||
let updated = repo.get_by_id(&id).await.ok().flatten();
|
||||
json_ok(json!({ "id": id, "project": updated }))
|
||||
}
|
||||
Ok(false) => CallToolResult::error(format!("项目不存在: {id}")),
|
||||
Ok(false) => CallToolResult::error("记录已被其他端修改,请重新获取最新数据后重试"),
|
||||
Err(e) => err_str(e),
|
||||
}
|
||||
})
|
||||
@@ -694,12 +697,14 @@ fn update_task(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> {
|
||||
created_at: existing.created_at,
|
||||
updated_at: now,
|
||||
};
|
||||
match repo.update_full(&rec).await {
|
||||
// 权威 CAS:以 get_by_id 刚读到的 updated_at 为 expected,单条原子条件写。
|
||||
// 封死「读 existing → 写 rec」之间的跨进程竞态窗口:并发端(GUI)已改则 affected==0。
|
||||
match repo.update_full_cas(&rec, &existing.updated_at).await {
|
||||
Ok(true) => {
|
||||
let updated = repo.get_by_id(&id).await.ok().flatten();
|
||||
json_ok(json!({ "id": id, "task": updated }))
|
||||
}
|
||||
Ok(false) => CallToolResult::error(format!("任务不存在: {id}")),
|
||||
Ok(false) => CallToolResult::error("记录已被其他端修改,请重新获取最新数据后重试"),
|
||||
Err(e) => err_str(e),
|
||||
}
|
||||
})
|
||||
@@ -854,12 +859,14 @@ fn update_idea(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> {
|
||||
created_at: existing.created_at,
|
||||
updated_at: now,
|
||||
};
|
||||
match repo.update_full(&rec).await {
|
||||
// 权威 CAS:以 get_by_id 刚读到的 updated_at 为 expected,单条原子条件写。
|
||||
// 封死「读 existing → 写 rec」之间的跨进程竞态窗口:并发端(GUI)已改则 affected==0。
|
||||
match repo.update_full_cas(&rec, &existing.updated_at).await {
|
||||
Ok(true) => {
|
||||
let updated = repo.get_by_id(&id).await.ok().flatten();
|
||||
json_ok(json!({ "id": id, "idea": updated }))
|
||||
}
|
||||
Ok(false) => CallToolResult::error(format!("想法不存在: {id}")),
|
||||
Ok(false) => CallToolResult::error("记录已被其他端修改,请重新获取最新数据后重试"),
|
||||
Err(e) => err_str(e),
|
||||
}
|
||||
})
|
||||
@@ -928,10 +935,14 @@ fn score_idea(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> {
|
||||
Err(e) => return CallToolResult::error(format!("评分序列化失败: {e}")),
|
||||
});
|
||||
rec.updated_at = now;
|
||||
if let Err(e) = repo.update_full(&rec).await {
|
||||
return err_str(e);
|
||||
// 权威 CAS:以 get_by_id 刚读到的 updated_at 为 expected,单条原子条件写。
|
||||
// 旧 update_full 忽略 Ok(false) 静默覆盖;改后并发端(GUI)已改则 affected==0,
|
||||
// 报版本冲突,不再互相覆盖。
|
||||
match repo.update_full_cas(&rec, &idea.updated_at).await {
|
||||
Ok(true) => json_ok(json!({ "id": id, "idea": rec, "scores": scores })),
|
||||
Ok(false) => CallToolResult::error("记录已被其他端修改,请重新获取最新数据后重试"),
|
||||
Err(e) => err_str(e),
|
||||
}
|
||||
json_ok(json!({ "id": id, "idea": rec, "scores": scores }))
|
||||
})
|
||||
}
|
||||
|
||||
@@ -973,8 +984,9 @@ fn run_workflow(_ctx: &Ctx, _args: Value) -> BoxFuture<'static, CallToolResult>
|
||||
// handler 实现 — 回收站
|
||||
// ============================================================
|
||||
|
||||
fn list_trash(ctx: &Ctx, _args: Value) -> BoxFuture<'static, CallToolResult> {
|
||||
fn list_trash(ctx: &Ctx, args: Value) -> BoxFuture<'static, CallToolResult> {
|
||||
let db = ctx.db.clone();
|
||||
let (offset, limit) = pagination(&args);
|
||||
Box::pin(async move {
|
||||
let projects = match ProjectRepo::new(&db).list_deleted().await {
|
||||
Ok(v) => v,
|
||||
@@ -984,11 +996,34 @@ fn list_trash(ctx: &Ctx, _args: Value) -> BoxFuture<'static, CallToolResult> {
|
||||
Ok(v) => v,
|
||||
Err(e) => return err_str(e),
|
||||
};
|
||||
// 分页:与 list_projects/tasks/ideas 同契约(offset/limit/has_more)。
|
||||
// projects 与 tasks 各自独立分页(同页语义:各自第 N 页),has_more 取两者并集。
|
||||
// 回收站数据量小(仅软删项),内存分页足够,避免给共享 list_deleted 加 SQL 分页侵入。
|
||||
// 探页:多取一条探测是否有下一页(对齐 list_* 的 limit+1 探页语义)。
|
||||
let project_probe: Vec<_> = projects
|
||||
.into_iter()
|
||||
.skip(offset as usize)
|
||||
.take(limit as usize + 1)
|
||||
.collect();
|
||||
let project_has_more = project_probe.len() > limit as usize;
|
||||
let project_page: Vec<_> = project_probe.into_iter().take(limit as usize).collect();
|
||||
let task_probe: Vec<_> = tasks
|
||||
.into_iter()
|
||||
.skip(offset as usize)
|
||||
.take(limit as usize + 1)
|
||||
.collect();
|
||||
let task_has_more = task_probe.len() > limit as usize;
|
||||
let task_page: Vec<_> = task_probe.into_iter().take(limit as usize).collect();
|
||||
json_ok(json!({
|
||||
"projects": projects,
|
||||
"tasks": tasks,
|
||||
"project_count": projects.len(),
|
||||
"task_count": tasks.len()
|
||||
"projects": project_page,
|
||||
"tasks": task_page,
|
||||
"project_count": project_page.len(),
|
||||
"task_count": task_page.len(),
|
||||
"offset": offset,
|
||||
"limit": limit,
|
||||
"has_more": project_has_more || task_has_more,
|
||||
"project_has_more": project_has_more,
|
||||
"task_has_more": task_has_more
|
||||
}))
|
||||
})
|
||||
}
|
||||
@@ -1820,4 +1855,52 @@ mod tests {
|
||||
assert!(visible_for_test(false, "search_knowledge"));
|
||||
assert!(visible_for_test(false, "insert_knowledge"));
|
||||
}
|
||||
|
||||
// ── list_trash 分页(offset/limit/has_more,projects/tasks 各自独立分页)──
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_trash_pagination_and_has_more() {
|
||||
let ctx = test_ctx().await;
|
||||
// 3 项目软删进回收站 + 1 任务软删(宿主项目不删,仅任务进回收站)
|
||||
let mut pids = Vec::new();
|
||||
for i in 0..3 {
|
||||
pids.push(seed_project(&ctx, &format!("回收项目{i}")).await);
|
||||
}
|
||||
for p in &pids {
|
||||
ProjectRepo::new(&ctx.db).soft_delete(p).await.unwrap();
|
||||
}
|
||||
let tid = seed_task(&ctx, &pids[0], "回收任务").await;
|
||||
TaskRepo::new(&ctx.db).soft_delete(&tid).await.unwrap();
|
||||
|
||||
// 首页 limit=2 → 项目 2 条(仍有下一页),任务 1 条(无下一页)
|
||||
let r = list_trash(&ctx, json!({ "limit": 2, "offset": 0 })).await;
|
||||
assert!(r.is_error.is_none(), "{:?}", text_of(&r));
|
||||
let v = json_of(&r);
|
||||
assert_eq!(v["projects"].as_array().unwrap().len(), 2);
|
||||
assert_eq!(v["tasks"].as_array().unwrap().len(), 1);
|
||||
assert_eq!(v["project_count"], 2);
|
||||
assert_eq!(v["task_count"], 1);
|
||||
assert_eq!(v["project_has_more"], true, "3 项目取 2 还有下一页");
|
||||
assert_eq!(v["task_has_more"], false);
|
||||
assert_eq!(v["has_more"], true);
|
||||
assert_eq!(v["limit"], 2);
|
||||
assert_eq!(v["offset"], 0);
|
||||
|
||||
// 第二页 offset=2 → 项目 1 条,两边都无下一页
|
||||
let r = list_trash(&ctx, json!({ "limit": 2, "offset": 2 })).await;
|
||||
let v = json_of(&r);
|
||||
assert_eq!(v["projects"].as_array().unwrap().len(), 1);
|
||||
assert_eq!(v["project_has_more"], false);
|
||||
assert_eq!(v["task_has_more"], false);
|
||||
assert_eq!(v["has_more"], false);
|
||||
}
|
||||
|
||||
/// limit 超上限钳到 100(对齐其他 list 工具的钳制)。
|
||||
#[tokio::test]
|
||||
async fn list_trash_caps_limit_to_100() {
|
||||
let ctx = test_ctx().await;
|
||||
let r = list_trash(&ctx, json!({ "limit": 999 })).await;
|
||||
assert!(r.is_error.is_none(), "{:?}", text_of(&r));
|
||||
assert_eq!(json_of(&r)["limit"], 100, "limit 超上限应钳到 100");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -538,6 +538,43 @@ impl IdeaRepo {
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 原子乐观更新(CAS 版 [`update_full`]):整体更新记录,但仅当 DB 当前 `updated_at`
|
||||
/// 与 `expected_updated_at` 一致时才写入(`WHERE id=? AND updated_at=?expected`)。
|
||||
///
|
||||
/// 关闭读-改-写跨进程 TOCTOU(devflow-mcp 多进程缺陷 P0-1):MCP 进程先 `get_by_id`
|
||||
/// 读 `existing.updated_at` 作 expected,再调本方法单条原子条件写。并发端(GUI 进程)
|
||||
/// 若已改动该记录,`affected==0` 返回 `false`,调用方据此报「记录已被其他端修改」
|
||||
/// 而非静默覆盖(旧 `update_full` 无条件覆盖,存在跨进程互相覆盖竞态)。
|
||||
///
|
||||
/// 不动 GUI 的无条件 [`update_full`](无 expected 语义,保持现状);本方法仅服务
|
||||
/// 需要乐观锁版本的调用方(df-mcp update_idea / score_idea)。
|
||||
/// `Ok(false)` = id 不存在或版本冲突。
|
||||
pub async fn update_full_cas(
|
||||
&self,
|
||||
record: &IdeaRecord,
|
||||
expected_updated_at: &str,
|
||||
) -> Result<bool> {
|
||||
let conn = self.conn.clone();
|
||||
let rec = record.clone();
|
||||
let expected = expected_updated_at.to_owned();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let affected = guard
|
||||
.execute(
|
||||
"UPDATE ideas SET title = ?1, description = ?2, status = ?3, priority = ?4, score = ?5, tags = ?6, source = ?7, promoted_to = ?8, ai_analysis = ?9, scores = ?10, related_ids = ?11, updated_at = ?12 WHERE id = ?13 AND updated_at = ?14",
|
||||
params![
|
||||
rec.title, rec.description, rec.status.as_str(), rec.priority,
|
||||
rec.score, rec.tags, rec.source, rec.promoted_to, rec.ai_analysis,
|
||||
rec.scores, rec.related_ids, rec.updated_at, rec.id, expected
|
||||
],
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
Ok(affected > 0)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
}
|
||||
|
||||
impl KnowledgeRepo {
|
||||
@@ -1497,4 +1534,50 @@ mod tests {
|
||||
// i1 活跃不出现;i3 删除最晚在前
|
||||
assert_eq!(ids, vec!["i3", "i2"], "list_deleted 应只含回收站灵感,按 updated_at DESC");
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// update_full_cas 原子乐观更新(devflow-mcp P0-1 TOCTOU 关闭)
|
||||
// 锁定:① expected 与 DB updated_at 一致 → 写入 true;② expected 旧版本(并发已改)
|
||||
// → 不写入返回 false 且原值保留(不覆盖他人修改)。
|
||||
// ============================================================
|
||||
|
||||
#[tokio::test]
|
||||
async fn idea_update_full_cas_success_when_expected_matches() {
|
||||
let repo = setup_idea_repo().await;
|
||||
repo.insert(irec("i1", "原标题")).await.unwrap();
|
||||
let current = repo.get_by_id("i1").await.unwrap().unwrap();
|
||||
// 本地构建新版本(title + updated_at 递增)
|
||||
let mut rec = current.clone();
|
||||
rec.title = "本地新标题".to_string();
|
||||
rec.updated_at = "1800000000000".to_string();
|
||||
let ok = repo
|
||||
.update_full_cas(&rec, ¤t.updated_at)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(ok, "expected 与 DB 一致应写入");
|
||||
let after = repo.get_by_id("i1").await.unwrap().unwrap();
|
||||
assert_eq!(after.title, "本地新标题");
|
||||
assert_eq!(after.updated_at, "1800000000000");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn idea_update_full_cas_conflict_when_expected_stale() {
|
||||
let repo = setup_idea_repo().await;
|
||||
repo.insert(irec("i1", "原标题")).await.unwrap();
|
||||
// 另一进程(如 GUI)先无条件覆盖:updated_at 从 1700 变 1800
|
||||
let mut other = repo.get_by_id("i1").await.unwrap().unwrap();
|
||||
other.title = "他人已改".to_string();
|
||||
other.updated_at = "1800000000000".to_string();
|
||||
assert!(repo.update_full(&other).await.unwrap());
|
||||
|
||||
// 本地持旧版本 expected(1700)→ CAS 应拒绝,不覆盖他人修改
|
||||
let mut mine = repo.get_by_id("i1").await.unwrap().unwrap();
|
||||
mine.title = "我的修改".to_string();
|
||||
mine.updated_at = "1900000000000".to_string();
|
||||
let ok = repo.update_full_cas(&mine, "1700000000000").await.unwrap();
|
||||
assert!(!ok, "expected 过期(并发已改)应返回 false");
|
||||
let after = repo.get_by_id("i1").await.unwrap().unwrap();
|
||||
assert_eq!(after.title, "他人已改", "CAS 冲突不得覆盖他人修改");
|
||||
assert_eq!(after.updated_at, "1800000000000");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -414,6 +414,41 @@ impl ProjectRepo {
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 原子乐观更新(CAS 版 [`update_full`]):整体更新记录,但仅当 DB 当前 `updated_at`
|
||||
/// 与 `expected_updated_at` 一致时才写入(`WHERE id=? AND updated_at=?expected`)。
|
||||
///
|
||||
/// 关闭读-改-写跨进程 TOCTOU(devflow-mcp 多进程缺陷 P0-1):MCP 进程先 `get_by_id`
|
||||
/// 读 `existing.updated_at` 作 expected,再调本方法单条原子条件写。并发端(GUI 进程)
|
||||
/// 若已改动该记录,`affected==0` 返回 `false`,调用方据此报「记录已被其他端修改」
|
||||
/// 而非静默覆盖(旧 `update_full` 无条件覆盖,存在跨进程互相覆盖竞态)。
|
||||
///
|
||||
/// 不动 GUI 的无条件 [`update_full`](无 expected 语义,保持现状);本方法仅服务
|
||||
/// 需要乐观锁版本的调用方(df-mcp update_project)。`Ok(false)` = id 不存在或版本冲突。
|
||||
pub async fn update_full_cas(
|
||||
&self,
|
||||
record: &ProjectRecord,
|
||||
expected_updated_at: &str,
|
||||
) -> Result<bool> {
|
||||
let conn = self.conn.clone();
|
||||
let rec = record.clone();
|
||||
let expected = expected_updated_at.to_owned();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let affected = guard
|
||||
.execute(
|
||||
"UPDATE projects SET name = ?1, description = ?2, status = ?3, idea_id = ?4, path = ?5, stack = ?6, updated_at = ?7 WHERE id = ?8 AND updated_at = ?9",
|
||||
params![
|
||||
rec.name, rec.description, rec.status.as_str(), rec.idea_id,
|
||||
rec.path, rec.stack, rec.updated_at, rec.id, expected
|
||||
],
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
Ok(affected > 0)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 彻底删除:事务级联删全部关联子表→projects(不可恢复)
|
||||
///
|
||||
/// SQLite 已开 PRAGMA foreign_keys=ON 但表无 ON DELETE CASCADE,ALTER 改不了 FK 约束,
|
||||
@@ -611,3 +646,74 @@ impl_repo!(
|
||||
)
|
||||
}
|
||||
);
|
||||
|
||||
// ============================================================
|
||||
// 单元测试 — update_full_cas 原子乐观更新(devflow-mcp P0-1 TOCTOU 关闭)
|
||||
// 锁定:① expected 与 DB updated_at 一致 → 写入 true;② expected 旧版本(并发已改)
|
||||
// → 不写入返回 false 且原值保留(不覆盖他人修改)。
|
||||
// ============================================================
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::db::Database;
|
||||
|
||||
fn prec(id: &str, name: &str) -> ProjectRecord {
|
||||
ProjectRecord {
|
||||
id: id.to_string(),
|
||||
name: name.to_string(),
|
||||
description: String::new(),
|
||||
status: ProjectStatus::Planning,
|
||||
idea_id: None,
|
||||
path: None,
|
||||
stack: None,
|
||||
created_at: "1700000000000".to_string(),
|
||||
updated_at: "1700000000000".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn setup() -> ProjectRepo {
|
||||
let db = Database::open_in_memory().await.expect("open_in_memory");
|
||||
ProjectRepo::new(&db)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_full_cas_success_when_expected_matches() {
|
||||
let repo = setup().await;
|
||||
repo.insert(prec("p1", "原项目")).await.unwrap();
|
||||
let current = repo.get_by_id("p1").await.unwrap().unwrap();
|
||||
// 本地构建新版本(name + updated_at 递增)
|
||||
let mut rec = current.clone();
|
||||
rec.name = "新项目".to_string();
|
||||
rec.updated_at = "1800000000000".to_string();
|
||||
let ok = repo
|
||||
.update_full_cas(&rec, ¤t.updated_at)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(ok, "expected 与 DB 一致应写入");
|
||||
let after = repo.get_by_id("p1").await.unwrap().unwrap();
|
||||
assert_eq!(after.name, "新项目");
|
||||
assert_eq!(after.updated_at, "1800000000000");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_full_cas_conflict_when_expected_stale() {
|
||||
let repo = setup().await;
|
||||
repo.insert(prec("p1", "原项目")).await.unwrap();
|
||||
// 另一进程(如 GUI)先无条件覆盖:updated_at 从 1700 变 1800
|
||||
let mut other = repo.get_by_id("p1").await.unwrap().unwrap();
|
||||
other.name = "他人已改".to_string();
|
||||
other.updated_at = "1800000000000".to_string();
|
||||
assert!(repo.update_full(&other).await.unwrap());
|
||||
|
||||
// 本地持旧版本 expected(1700)→ CAS 应拒绝,不覆盖他人修改
|
||||
let mut mine = repo.get_by_id("p1").await.unwrap().unwrap();
|
||||
mine.name = "我的修改".to_string();
|
||||
mine.updated_at = "1900000000000".to_string();
|
||||
let ok = repo.update_full_cas(&mine, "1700000000000").await.unwrap();
|
||||
assert!(!ok, "expected 过期(并发已改)应返回 false");
|
||||
let after = repo.get_by_id("p1").await.unwrap().unwrap();
|
||||
assert_eq!(after.name, "他人已改", "CAS 冲突不得覆盖他人修改");
|
||||
assert_eq!(after.updated_at, "1800000000000");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -743,6 +743,44 @@ impl TaskRepo {
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
|
||||
/// 原子乐观更新(CAS 版 [`update_full`]):整体更新记录,但仅当 DB 当前 `updated_at`
|
||||
/// 与 `expected_updated_at` 一致时才写入(`WHERE id=? AND updated_at=?expected`)。
|
||||
///
|
||||
/// 关闭读-改-写跨进程 TOCTOU(devflow-mcp 多进程缺陷 P0-1):MCP 进程先 `get_by_id`
|
||||
/// 读 `existing.updated_at` 作 expected,再调本方法单条原子条件写。并发端(GUI 进程)
|
||||
/// 若已改动该记录,`affected==0` 返回 `false`,调用方据此报「记录已被其他端修改」
|
||||
/// 而非静默覆盖。对齐 [`advance_status_atomic`](同款 `WHERE ... = ?expected` 原子条件写)。
|
||||
///
|
||||
/// 不动 GUI 的无条件 [`update_full`](无 expected 语义,保持现状);本方法仅服务
|
||||
/// 需要乐观锁版本的调用方(df-mcp update_task)。`Ok(false)` = id 不存在或版本冲突。
|
||||
pub async fn update_full_cas(
|
||||
&self,
|
||||
record: &TaskRecord,
|
||||
expected_updated_at: &str,
|
||||
) -> Result<bool> {
|
||||
let conn = self.conn.clone();
|
||||
let rec = record.clone();
|
||||
let expected = expected_updated_at.to_owned();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let guard = conn.blocking_lock();
|
||||
let affected = guard
|
||||
.execute(
|
||||
"UPDATE tasks SET project_id = ?1, title = ?2, description = ?3, status = ?4, priority = ?5, branch_name = ?6, assignee = ?7, workflow_def_id = ?8, base_branch = ?9, review_rounds = ?10, output_json = ?11, idea_id = ?12, queue = ?13, parent_id = ?14, content_json = ?15, module_id = ?16, updated_at = ?17 WHERE id = ?18 AND updated_at = ?19",
|
||||
params![
|
||||
rec.project_id, rec.title, rec.description, rec.status.as_str(), rec.priority,
|
||||
rec.branch_name, rec.assignee, rec.workflow_def_id, rec.base_branch,
|
||||
rec.review_rounds, rec.output_json, rec.idea_id,
|
||||
rec.queue, rec.parent_id, rec.content_json, rec.module_id,
|
||||
rec.updated_at, rec.id, expected
|
||||
],
|
||||
)
|
||||
.map_err(storage_err)?;
|
||||
Ok(affected > 0)
|
||||
})
|
||||
.await
|
||||
.map_err(storage_err)?
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
@@ -1160,4 +1198,50 @@ mod tests {
|
||||
assert_eq!(updated.queue, "todo");
|
||||
assert_eq!(updated.status.as_str(), "todo");
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// update_full_cas 原子乐观更新(devflow-mcp P0-1 TOCTOU 关闭)
|
||||
// 锁定:① expected 与 DB updated_at 一致 → 写入 true;② expected 旧版本(并发已改)
|
||||
// → 不写入返回 false 且原值保留(不覆盖他人修改)。
|
||||
// ============================================================
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_full_cas_success_when_expected_matches() {
|
||||
let repo = setup().await;
|
||||
repo.insert(trec("t1", "todo", None)).await.unwrap();
|
||||
let current = repo.get_by_id("t1").await.unwrap().unwrap();
|
||||
// 本地构建新版本(title + updated_at 递增)
|
||||
let mut rec = current.clone();
|
||||
rec.title = "本地新标题".to_string();
|
||||
rec.updated_at = "1800000000000".to_string();
|
||||
let ok = repo
|
||||
.update_full_cas(&rec, ¤t.updated_at)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(ok, "expected 与 DB 一致应写入");
|
||||
let after = repo.get_by_id("t1").await.unwrap().unwrap();
|
||||
assert_eq!(after.title, "本地新标题");
|
||||
assert_eq!(after.updated_at, "1800000000000");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_full_cas_conflict_when_expected_stale() {
|
||||
let repo = setup().await;
|
||||
repo.insert(trec("t1", "todo", None)).await.unwrap();
|
||||
// 另一进程(如 GUI)先无条件覆盖:updated_at 从 1700 变 1800
|
||||
let mut other = repo.get_by_id("t1").await.unwrap().unwrap();
|
||||
other.title = "他人已改".to_string();
|
||||
other.updated_at = "1800000000000".to_string();
|
||||
assert!(repo.update_full(&other).await.unwrap());
|
||||
|
||||
// 本地持旧版本 expected(1700)→ CAS 应拒绝,不覆盖他人修改
|
||||
let mut mine = repo.get_by_id("t1").await.unwrap().unwrap();
|
||||
mine.title = "我的修改".to_string();
|
||||
mine.updated_at = "1900000000000".to_string();
|
||||
let ok = repo.update_full_cas(&mine, "1700000000000").await.unwrap();
|
||||
assert!(!ok, "expected 过期(并发已改)应返回 false");
|
||||
let after = repo.get_by_id("t1").await.unwrap().unwrap();
|
||||
assert_eq!(after.title, "他人已改", "CAS 冲突不得覆盖他人修改");
|
||||
assert_eq!(after.updated_at, "1800000000000");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user