lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化: 存储层清债与子轨迹来源迁移

Browse Source 
- V44 加 ai_conversations.origin 列(子代理轨迹来源标记)
- V21 断点续迁改造(逐对话进度判定,崩溃重跑补漏)
- 手写事务改 guard.transaction,panic 不残留打开事务
- 软删 CAS 补 deleted_at IS NULL(任务/项目/灵感三处)
- 父任务校验拒回收站挂子,删模块先清依赖边
- 列缺失与未知 status 降级补 warn 留痕,limit=0 返空集
This commit is contained in:
lxy
2026-10-03 15:34:55 +08:00
parent 3f4c914dbf
commit 499aa49e49
10 changed files with 796 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/df-storage/src/crud/conversation_repo.rs
+88 -35
View File
@@ -40,8 +40,31 @@ fn ai_provider_from_row(row: &Row<'_>) -> std::result::Result<AiProviderRecord,
.unwrap_or_default(),
}
};
// id 先读:① 结构体字段必填列(缺失直接 ? 报错);② enabled/weight 降级时日志带行标识,
// 坏数据可定位到具体行而非只剩全局报警。
let id: String = row.get("id")?;
// enabled/weight 列老库经 v19 迁移补建,DEFAULT 1 / DEFAULT 50。get 失败(列缺失
// 或 NULL 坏数据)降级默认值保功能(防单行坏数据拖垮 list_all),但 warn 留痕:
// 正常新库列必然存在,此分支触发即迁移未跑/表结构被外力改动,坏数据可观测。
// from_row 按 i32 取列值兼容(SQLite 无真 BOOLEAN),0→false/非0→true。
let enabled = match row.get::<_, i32>("enabled") {
Ok(v) => v != 0,
Err(e) => {
tracing::warn!("ai_providers[{id}] enabled 列读取失败({e}),降级默认 enabled=true");
true
}
};
// weight 读侧 clamp [0,100]:与 insert/update_full 落库的 `.min(100)` 对齐,
// 防老库(clamp 落地前写入的)或外部直改 DB 产生的越界值污染路由权重语义。
let weight = match row.get::<_, i32>("weight") {
Ok(v) => v.clamp(0, 100) as u32,
Err(e) => {
tracing::warn!("ai_providers[{id}] weight 列读取失败({e}),降级默认 weight=50");
50
}
};
Ok(AiProviderRecord {
id: row.get("id")?,
id,
name: row.get("name")?,
provider_type: row.get("provider_type")?,
api_key: row.get("api_key")?,
@@ -53,14 +76,8 @@ fn ai_provider_from_row(row: &Row<'_>) -> std::result::Result<AiProviderRecord,
config: row.get("config")?,
created_at: row.get("created_at")?,
updated_at: row.get("updated_at")?,
// enabled/weight 列老库经 v19 迁移补建,DEFAULT 1 / DEFAULT 50。
// from_row 按 i32 取列值兼容(SQLite 无真 BOOLEAN),0→false/非0→true。
enabled: row.get::<_, i32>("enabled").unwrap_or(1) != 0,
// weight 读侧 clamp [0,100]:与 insert/update_full 落库的 `.min(100)` 对齐,
// 防老库(clamp 落地前写入的)或外部直改 DB 产生的越界值污染路由权重语义。
weight: row.get::<_, i32>("weight")
.unwrap_or(50)
.clamp(0, 100) as u32,
enabled,
weight,
})
}
@@ -88,6 +105,9 @@ fn ai_conversation_from_row(row: &Row<'_>) -> std::result::Result<AiConversation
pending_approvals: row.get("pending_approvals")?,
created_at: row.get("created_at")?,
updated_at: row.get("updated_at")?,
// origin 列老库经 v44 迁移补建。unwrap_or(None) 兜底(同 message_id v21 模式):
// 迁移未跑/极端情况防御,NULL=主会话。
origin: row.get("origin").unwrap_or(None),
})
}
@@ -163,13 +183,13 @@ impl_repo!(
from_row => |row| ai_conversation_from_row(row),
insert => |conn, rec| {
conn.execute(
"INSERT INTO ai_conversations (id, title, messages, provider_id, model, models, archived, pinned, prompt_tokens, completion_tokens, pinned_goals, pending_approvals, created_at, updated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
"INSERT INTO ai_conversations (id, title, messages, provider_id, model, models, archived, pinned, prompt_tokens, completion_tokens, pinned_goals, pending_approvals, created_at, updated_at, origin)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)",
params![
rec.id, rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived,
if rec.pinned { 1i32 } else { 0i32 },
rec.prompt_tokens, rec.completion_tokens,
rec.pinned_goals, rec.pending_approvals, rec.created_at, rec.updated_at
rec.pinned_goals, rec.pending_approvals, rec.created_at, rec.updated_at, rec.origin
],
)
},
@@ -261,33 +281,28 @@ impl AiToolExecutionRepo {
return Ok(());
}
let conn = self.conn.clone();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
guard.execute_batch("BEGIN").map_err(storage_err)?;
let result = (|| -> std::result::Result<(), rusqlite::Error> {
let mut stmt = guard.prepare(
tokio::task::spawn_blocking(move || -> Result<()> {
let mut guard = conn.blocking_lock();
// transaction() guard 范式(同 clear_conversation_atomic / message_repo::insert_batch):
// rusqlite Transaction 在 Drop 未 commit 时自动回滚,中途 panic/早退不残留打开事务;
// 原手写 BEGIN/COMMIT 无此保证,panic 后连接带开事务,后续写全部静默失败。
let tx = guard.transaction().map_err(storage_err)?;
{
let mut stmt = tx.prepare(
"INSERT INTO ai_tool_executions (id, conversation_id, message_id, tool_call_id, tool_name, arguments, result, status, risk_level, requested_at, executed_at, decided_by)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
)?;
)
.map_err(storage_err)?;
for rec in &records {
stmt.execute(params![
rec.id, rec.conversation_id, rec.message_id, &*rec.tool_call_id, rec.tool_name,
rec.arguments, rec.result, rec.status, rec.risk_level,
rec.requested_at, rec.executed_at, rec.decided_by
])?;
}
Ok(())
})();
match result {
Ok(()) => {
guard.execute_batch("COMMIT").map_err(storage_err)?;
}
Err(e) => {
// 回滚失败静默(尽量保一致性;rollback 失败通常是连接已坏,交给上层)
let _ = guard.execute_batch("ROLLBACK");
return Err(storage_err(e));
])
.map_err(storage_err)?;
}
}
tx.commit().map_err(storage_err)?;
Ok(())
})
.await
@@ -606,6 +621,12 @@ impl AiConversationRepo {
/// 参数化绑定(替代原调用方的 format! 拼 SQL + execute_batch),防 snapshot 含引号/
/// 特殊字符致注入或损坏。列对齐 conversation_checkpoints(id, conv_id, snapshot,
/// token_total, created_at)。
///
/// 冲突语义:调用方 ck_id 由 (conv, 消息数) 确定性生成,同轮重复 save 撞同 id 是
/// 预期幂等跳过;但静默会掩盖另一情形——会话回退/重生成回同消息数且内容已变,
/// 新 snapshot 被旧 checkpoint 挡住未落库。affected=0 时 warn 留痕,两种情形都
/// 可观测(是否异常由调用方日志上下文判别)。签名保 Result<()> 不外泄跳过信息:
/// 跳过非错误,调用方对 Err 已按 warn 处理,无需改调用点。
pub async fn insert_checkpoint(
&self,
id: &str,
@@ -619,20 +640,25 @@ impl AiConversationRepo {
let conv_id = conv_id.to_string();
let snapshot = snapshot.to_string();
let created_at = created_at.to_string();
let _ = tokio::task::spawn_blocking(move || {
tokio::task::spawn_blocking(move || -> Result<()> {
let guard = conn.blocking_lock();
guard
let affected = guard
.execute(
"INSERT OR IGNORE INTO conversation_checkpoints \
(id, conv_id, snapshot, token_total, created_at) \
VALUES (?1, ?2, ?3, ?4, ?5)",
params![id, conv_id, snapshot, token_total, created_at],
)
.map_err(storage_err)
.map_err(storage_err)?;
if affected == 0 {
tracing::warn!(
"checkpoint 幂等跳过(同 id 已存在,新 snapshot 未落库): id={id} conv={conv_id}"
);
}
Ok(())
})
.await
.map_err(storage_err)??;
Ok(())
.map_err(storage_err)?
}
/// 清空对话消息内容(保留 conversation 记录本身,只清 messages JSON + 清零 token 计数)
@@ -943,4 +969,31 @@ mod tests {
let got2 = repo.find_by_conv_tool_args("conv1", "write_file", "{\"path\":\"a.go\"}").await.expect("query");
assert_eq!(got2.unwrap().id, "a2", "同三元组应返回最新落定记录");
}
/// insert_checkpoint 冲突语义:同 id 二次写入 Ok 不报错(幂等跳过,affected=0 走 warn),
/// 保留首次 snapshot。锁死 OR IGNORE 行为,防未来误改成 REPLACE 静默覆盖快照。
#[tokio::test]
async fn insert_checkpoint_conflict_skips_and_keeps_first() {
let db = Database::open_in_memory().await.expect("open_in_memory");
let repo = AiConversationRepo::new(&db);
repo.insert_checkpoint("ck_1", "conv1", r#"[{"seq":0}]"#, 1, "100")
.await
.expect("首次写入");
// 同 id 不同内容:IGNORE 生效应 Ok,首次 snapshot 保留
repo.insert_checkpoint("ck_1", "conv1", r#"[{"seq":0},{"seq":1}]"#, 2, "200")
.await
.expect("冲突写入应幂等跳过不报错");
let conn = db.conn();
let g = conn.lock().await;
let count: i64 = g
.query_row("SELECT COUNT(*) FROM conversation_checkpoints WHERE id = 'ck_1'", [], |r| r.get(0))
.expect("count");
assert_eq!(count, 1, "同 id 只应有一行");
let snapshot: String = g
.query_row("SELECT snapshot FROM conversation_checkpoints WHERE id = 'ck_1'", [], |r| r.get(0))
.expect("snapshot");
assert_eq!(snapshot, r#"[{"seq":0}]"#, "冲突时应保留首次快照");
}
}