fix: session pool 锁范围收敛与 replace_todos 原子性

- SessionPool.ensure_session_internal 改为 double-checked locking:
  先短暂持锁检查存在性,释放锁后执行耗时的 session 创建(含配置加载、
  agent 工厂构造),再次持锁插入并处理竞态。避免跨 session_factory.create().await
  持有全局锁导致所有 channel 的 session 访问串行化。
- storage::replace_todos 用 transaction() 包裹 DELETE + INSERT,保证原子性:
  中途失败自动回滚,避免 todos 列表丢失且无法恢复。事务内复用同一连接
  查询返回值,消除 drop(conn) 后重新 pool.get() 的冗余。
This commit is contained in:
oudecheng 2026-07-31 08:23:13 +08:00
parent 92db80dc3f
commit 7f05545488
2 changed files with 68 additions and 20 deletions

View File

@ -54,33 +54,52 @@ impl SessionPool {
}
/// 内部方法:创建 Session根据 is_scheduler 选择存储位置)
///
/// 使用 double-checked locking先短暂持锁检查存在性释放锁后执行耗时的
/// session 创建含配置加载、agent 工厂构造),再次持锁插入并处理竞态。
/// 避免跨 `session_factory.create().await` 持有全局锁导致所有 channel 的
/// session 访问串行化。
async fn ensure_session_internal(&self, channel_name: &str, is_scheduler: bool) -> Result<(), AgentError> {
let mut inner = self.inner.lock().await;
// 选择对应的存储
// Fast path: 已存在直接返回(短暂持锁)
{
let inner = self.inner.lock().await;
let sessions = if is_scheduler {
&mut inner.scheduler_sessions
&inner.scheduler_sessions
} else {
&mut inner.sessions
&inner.sessions
};
// 简化:只检查 session 是否存在,不做超时判断
if sessions.contains_key(channel_name) {
return Ok(());
}
}
// Session 不存在则创建
// Slow path: 锁外创建 session耗时操作加载配置、构造 agent 工厂)
let (user_tx, _rx) = mpsc::channel::<WsOutbound>(100);
let session = self
.session_factory
.create(channel_name.to_string(), user_tx)
.await?;
// 再次持锁插入,处理竞态(另一个并发任务可能已插入)
let mut inner = self.inner.lock().await;
let sessions = if is_scheduler {
&mut inner.scheduler_sessions
} else {
&mut inner.sessions
};
if sessions.contains_key(channel_name) {
// 竞态:另一任务先插入,丢弃我们创建的 session
// drop session 释放资源user_tx 也 drop无泄漏
tracing::debug!(
channel = %channel_name,
"Session created concurrently by another task, discarding duplicate"
);
return Ok(());
}
sessions.insert(channel_name.to_string(), Arc::new(Mutex::new(session)));
inner
.session_timestamps
.insert(channel_name.to_string(), Instant::now());
Ok(())
}

View File

@ -1610,18 +1610,22 @@ impl SessionStore {
scope_key: &str,
items: &[TodoRecord],
) -> Result<Vec<TodoRecord>, StorageError> {
let conn = self.pool.get()?;
let mut conn = self.pool.get()?;
// 用 transaction()(非 unchecked_transaction保证严格事务语义
// 用户数据替换需保证原子性——中途失败必须回滚,避免 DELETE 后 INSERT
// 异常导致 todos 列表丢失且无法恢复。
let tx = conn.transaction()?;
let now = current_timestamp();
// Delete existing todos for this scope_key
conn.execute(
tx.execute(
"DELETE FROM todos WHERE scope_key = ?1",
params![scope_key],
)?;
// Insert new todos
for item in items {
conn.execute(
tx.execute(
"INSERT OR REPLACE INTO todos (id, scope_key, session_id, topic_id, content, status, priority, created_at, updated_at, created_by_message_id)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
params![
@ -1639,9 +1643,34 @@ impl SessionStore {
)?;
}
drop(conn);
self.list_todos(scope_key)
// 事务内复用同一连接查询返回值,避免 drop(conn) 后重新 pool.get()。
let mut stmt = tx.prepare(
"SELECT id, scope_key, session_id, topic_id, content, status, priority, created_at, updated_at, created_by_message_id
FROM todos
WHERE scope_key = ?1
ORDER BY created_at ASC",
)?;
let rows = stmt.query_map(params![scope_key], |row| {
Ok(TodoRecord {
id: row.get(0)?,
scope_key: row.get(1)?,
session_id: row.get(2)?,
topic_id: row.get(3)?,
content: row.get(4)?,
status: row.get(5)?,
priority: row.get(6)?,
created_at: row.get(7)?,
updated_at: row.get(8)?,
created_by_message_id: row.get(9)?,
})
})?;
let mut result = Vec::new();
for row in rows {
result.push(row?);
}
drop(stmt); // 释放 stmt 借用,才能 commit
tx.commit()?;
Ok(result)
}
pub fn list_todos(&self, scope_key: &str) -> Result<Vec<TodoRecord>, StorageError> {