From 7f055454884d7d95c4af0b90541c2d484c04463a Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Fri, 31 Jul 2026 08:23:13 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20session=20pool=20=E9=94=81=E8=8C=83?= =?UTF-8?q?=E5=9B=B4=E6=94=B6=E6=95=9B=E4=B8=8E=20replace=5Ftodos=20?= =?UTF-8?q?=E5=8E=9F=E5=AD=90=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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() 的冗余。 --- src/gateway/session_pool.rs | 47 ++++++++++++++++++++++++++----------- src/storage/mod.rs | 41 +++++++++++++++++++++++++++----- 2 files changed, 68 insertions(+), 20 deletions(-) diff --git a/src/gateway/session_pool.rs b/src/gateway/session_pool.rs index 546f57a..762202b 100644 --- a/src/gateway/session_pool.rs +++ b/src/gateway/session_pool.rs @@ -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; - - // 选择对应的存储 - let sessions = if is_scheduler { - &mut inner.scheduler_sessions - } else { - &mut inner.sessions - }; - - // 简化：只检查 session 是否存在，不做超时判断 - if sessions.contains_key(channel_name) { - return Ok(()); + // Fast path: 已存在直接返回（短暂持锁） + { + let inner = self.inner.lock().await; + let sessions = if is_scheduler { + &inner.scheduler_sessions + } else { + &inner.sessions + }; + if sessions.contains_key(channel_name) { + return Ok(()); + } } - // Session 不存在则创建 + // Slow path: 锁外创建 session（耗时操作：加载配置、构造 agent 工厂） let (user_tx, _rx) = mpsc::channel::(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(()) } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 4ed6816..a62cd34 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -1610,18 +1610,22 @@ impl SessionStore { scope_key: &str, items: &[TodoRecord], ) -> Result, 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, StorageError> {