From 4517e4a7249e69e859ca24fecd46382b63ac159c Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Mon, 17 Aug 2026 09:54:54 +0800 Subject: [PATCH] =?UTF-8?q?perf(db):=20=E7=83=AD=E8=B7=AF=E5=BE=84?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=20SQLite=20=E6=93=8D=E4=BD=9C=E6=94=B9=20spa?= =?UTF-8?q?wn=5Fblocking=20=E5=B9=B6=E5=AE=9A=E5=90=91=E6=9F=A5=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - processor/session 热点 DB 调用放到 blocking 线程池，避免阻塞 tokio worker - 定向查询 topic 首条用户消息（DB 侧 LIMIT 1），不再全量加载历史 - 新增 first_user_message_content 访问方法 --- src/gateway/processor.rs | 72 +++++++++++++++++++++++++--------- src/gateway/session.rs | 83 ++++++++++++++++++++++++++++++---------- src/storage/mod.rs | 27 +++++++++++++ src/storage/ports.rs | 8 ++++ 4 files changed, 150 insertions(+), 40 deletions(-) diff --git a/src/gateway/processor.rs b/src/gateway/processor.rs index 8439bd5..e31f0a2 100644 --- a/src/gateway/processor.rs +++ b/src/gateway/processor.rs @@ -450,7 +450,17 @@ impl InboundProcessor { // 异步生成 topic 描述（仅当描述为空且没有正在进行的生成任务时触发） if let Some(ref topic_id) = current_topic { let store = self.session_manager.store(); - if let Ok(Some(topic)) = store.get_topic(topic_id) + // SQLite 是同步 I/O：放到 blocking 线程池，避免阻塞 tokio worker + let store_for_lookup = store.clone(); + let topic_id_for_lookup = topic_id.clone(); + let topic_row = tokio::task::spawn_blocking(move || { + store_for_lookup.get_topic(&topic_id_for_lookup) + }) + .await + .ok() + .and_then(|r| r.ok()) + .unwrap_or(None); + if let Some(topic) = topic_row && (topic.description.is_none() || topic .description @@ -476,12 +486,17 @@ impl InboundProcessor { let in_flight = self.description_generation_in_flight.clone(); tokio::spawn(async move { - // 从 DB 查询该 topic 的第一条用户消息作为描述生成的依据 - let first_user_message = store_clone - .load_messages_for_topic_full(&topic_id_clone, None) - .ok() - .and_then(|msgs| msgs.into_iter().find(|m| m.role == "user")) - .map(|m| m.content); + // 定向查询该 topic 的第一条用户消息（DB 侧 LIMIT 1， + // 不再全量加载整个话题历史），并放到 blocking 线程池执行 + let store_for_query = store_clone.clone(); + let topic_id_for_query = topic_id_clone.clone(); + let first_user_message = tokio::task::spawn_blocking(move || { + store_for_query.first_user_message_content(&topic_id_for_query) + }) + .await + .ok() + .and_then(|r| r.ok()) + .unwrap_or(None); let message_content = match first_user_message { Some(content) => content, @@ -501,13 +516,27 @@ impl InboundProcessor { .await { Ok(description) => { - if let Err(e) = store_clone.update_topic_description( - &topic_id_clone, - &description, - ) { - tracing::error!(error = %e, topic_id = %topic_id_clone, "Failed to update topic description"); - } else { - tracing::info!(topic_id = %topic_id_clone, description = %description, "Topic description generated"); + let store_for_update = store_clone.clone(); + let topic_id_for_update = topic_id_clone.clone(); + let description_for_update = description.clone(); + let update_result = + tokio::task::spawn_blocking(move || { + store_for_update.update_topic_description( + &topic_id_for_update, + &description_for_update, + ) + }) + .await; + match update_result { + Ok(Ok(())) => { + tracing::info!(topic_id = %topic_id_clone, description = %description, "Topic description generated"); + } + Ok(Err(e)) => { + tracing::error!(error = %e, topic_id = %topic_id_clone, "Failed to update topic description"); + } + Err(e) => { + tracing::error!(error = %e, topic_id = %topic_id_clone, "Topic description update task panicked"); + } } } Err(e) => { @@ -570,11 +599,16 @@ impl InboundProcessor { // 恢复路径：下一条用户消息触发新的 process_one → 加载 history → // LLM 看到 "running" 占位 → 调用 wait_for_subagents → 消费 sub_done_q 结果。 let has_pending_subagents = if let Some(ref topic_id) = current_topic { - let pending = self - .session_manager - .store() - .list_pending_subagents(topic_id, Some("running")) - .unwrap_or_default(); + // SQLite 是同步 I/O：放到 blocking 线程池，避免阻塞 tokio worker + let store = self.session_manager.store(); + let topic_id_for_query = topic_id.clone(); + let pending = tokio::task::spawn_blocking(move || { + store.list_pending_subagents(&topic_id_for_query, Some("running")) + }) + .await + .ok() + .and_then(|r| r.ok()) + .unwrap_or_default(); if !pending.is_empty() { tracing::debug!( topic_id = %topic_id, diff --git a/src/gateway/session.rs b/src/gateway/session.rs index 34013f9..1177732 100644 --- a/src/gateway/session.rs +++ b/src/gateway/session.rs @@ -140,9 +140,29 @@ impl EmittedMessageHandler for BusToolCallEmitter { } } - // 拦截 todo_write 结果：即时持久化到 SQLite + // 拦截 todo_write 结果：即时持久化到 SQLite。 + // SQLite 是同步 I/O：放到 blocking 线程池，避免阻塞 tokio worker； + // await 保持与先前同步实现一致的顺序语义（同一 emitter 的连续 + // todo_write 不会乱序覆盖）。 if message.tool_name.as_deref() == Some("todo_write") { - self.persist_todo_write_result(&message); + let store = self.store.clone(); + let channel_name = self.channel_name.clone(); + let chat_id = self.chat_id.clone(); + let metadata = self.metadata.clone(); + let message_clone = message.clone(); + if let Err(e) = tokio::task::spawn_blocking(move || { + Self::persist_todo_write_result_sync( + &store, + &channel_name, + &chat_id, + &metadata, + &message_clone, + ) + }) + .await + { + tracing::warn!(error = %e, "todo_write persistence task failed"); + } } } @@ -195,8 +215,14 @@ impl EmittedMessageHandler for BusToolCallEmitter { } impl BusToolCallEmitter { - /// 从 todo_write 工具结果中提取 todos 并持久化 - fn persist_todo_write_result(&self, message: &ChatMessage) { + /// 从 todo_write 工具结果中提取 todos 并持久化（同步实现，供 spawn_blocking 调用） + fn persist_todo_write_result_sync( + store: &Arc, + channel_name: &str, + chat_id: &str, + metadata: &HashMap, + message: &ChatMessage, + ) { let parsed: serde_json::Value = match serde_json::from_str(&message.content) { Ok(v) => v, Err(_) => return, @@ -206,20 +232,15 @@ impl BusToolCallEmitter { return; }; - let session_id = crate::storage::persistent_session_id(&self.channel_name, &self.chat_id); + let session_id = crate::storage::persistent_session_id(channel_name, chat_id); // 优先用 topic_id（与 list_todos handler 和 tool 内存状态保持一致） - let scope_key = self - .metadata + let scope_key = metadata .get("topic_id") .filter(|t| !t.is_empty()) .cloned() .unwrap_or_else(|| session_id.clone()); - let topic_id = self - .metadata - .get("topic_id") - .filter(|t| !t.is_empty()) - .cloned(); + let topic_id = metadata.get("topic_id").filter(|t| !t.is_empty()).cloned(); let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -227,7 +248,7 @@ impl BusToolCallEmitter { .as_secs() as i64; // 读取现有 DB 记录，独立对比决定 created_by_message_id 是否更新 - let existing = self.store.list_todos(&scope_key).unwrap_or_default(); + let existing = store.list_todos(&scope_key).unwrap_or_default(); let existing_map: std::collections::HashMap<&str, &crate::storage::TodoRecord> = existing.iter().map(|r| (r.id.as_str(), r)).collect(); @@ -274,7 +295,7 @@ impl BusToolCallEmitter { "BusToolCallEmitter: persisting todo_write result" ); - if let Err(e) = self.store.replace_todos(&scope_key, &records) { + if let Err(e) = store.replace_todos(&scope_key, &records) { tracing::warn!(error = %e, %scope_key, "Failed to persist todo list from BusToolCallEmitter"); } } @@ -1132,10 +1153,17 @@ impl SessionManager { // 如果内存中没有当前话题，从数据库恢复最近活跃的话题 if guard.current_topic(chat_id).is_none() { let session_id = guard.persistent_session_id(chat_id); - let topics = self - .store - .list_topics(&session_id) - .map_err(|e| AgentError::Other(format!("Failed to list topics: {}", e)))?; + // SQLite 是同步 I/O：放到 blocking 线程池，避免阻塞 tokio worker。 + // session 互斥锁继续持有（tokio Mutex 允许跨 await），保证恢复/创建 + // 话题的原子性语义不变。 + let store_for_query = self.store.clone(); + let session_id_for_query = session_id.clone(); + let topics = tokio::task::spawn_blocking(move || { + store_for_query.list_topics(&session_id_for_query) + }) + .await + .map_err(|e| AgentError::Other(format!("Topic query task failed: {}", e)))? + .map_err(|e| AgentError::Other(format!("Failed to list topics: {}", e)))?; if let Some(latest_topic) = topics.first() { // 设置最近活跃的话题为当前话题 @@ -1149,8 +1177,14 @@ impl SessionManager { } else { // 数据库中也没有话题，自动创建默认话题 let title = format!("话题 {}", chrono::Local::now().format("%m/%d %H:%M")); - match self.store.create_topic(&session_id, &title, None) { - Ok(topic) => { + let store_for_create = self.store.clone(); + let session_id_for_create = session_id.clone(); + let create_result = tokio::task::spawn_blocking(move || { + store_for_create.create_topic(&session_id_for_create, &title, None) + }) + .await; + match create_result { + Ok(Ok(topic)) => { guard.set_current_topic(chat_id, Some(topic.id.clone())); tracing::info!( chat_id = %chat_id, @@ -1160,13 +1194,20 @@ impl SessionManager { "Auto-created default topic for new chat" ); } - Err(e) => { + Ok(Err(e)) => { tracing::error!( error = %e, session_id = %session_id, "Failed to auto-create default topic" ); } + Err(e) => { + tracing::error!( + error = %e, + session_id = %session_id, + "Topic creation task failed" + ); + } } } } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 32f8f08..408fe2e 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -1785,6 +1785,33 @@ impl SessionStore { } } + /// 定向查询指定话题的第一条 user 消息内容。 + /// + /// 数据库侧 `LIMIT 1`，避免为取单条消息全量加载并反序列化整个话题历史 + /// （话题越长，全量加载的 CPU/内存浪费越大）。 + pub fn first_user_message_content( + &self, + topic_id: &str, + ) -> Result, StorageError> { + let conn = self.pool.get()?; + let mut stmt = conn.prepare( + " + SELECT content + FROM messages + WHERE topic_id = ?1 AND role = 'user' + AND (system_context IS NULL OR system_context NOT LIKE 'history_compaction%') + ORDER BY seq ASC + LIMIT 1 + ", + )?; + let mut rows = stmt.query_map(params![topic_id], |row| row.get::<_, String>(0))?; + match rows.next() { + Some(Ok(content)) => Ok(Some(content)), + Some(Err(e)) => Err(e.into()), + None => Ok(None), + } + } + /// 获取指定话题的消息数量。 /// /// 使用 `SELECT COUNT(*)` 在数据库侧计数，避免将所有消息 diff --git a/src/storage/ports.rs b/src/storage/ports.rs index e5cd658..edf8dee 100644 --- a/src/storage/ports.rs +++ b/src/storage/ports.rs @@ -36,6 +36,10 @@ pub trait ConversationRepository: Send + Sync + 'static { session_id: Option<&str>, ) -> Result, StorageError>; + /// 定向查询指定话题的第一条 user 消息内容（数据库侧 LIMIT 1）。 + /// 避免为取单条消息而全量加载并反序列化整个话题历史。 + fn first_user_message_content(&self, topic_id: &str) -> Result, StorageError>; + fn append_message(&self, session_id: &str, message: &ChatMessage) -> Result<(), StorageError>; fn append_message_with_topic( @@ -298,6 +302,10 @@ impl ConversationRepository for super::SessionStore { super::SessionStore::load_messages_for_topic_full(self, topic_id, session_id) } + fn first_user_message_content(&self, topic_id: &str) -> Result, StorageError> { + super::SessionStore::first_user_message_content(self, topic_id) + } + fn compact_topic_history( &self, session_id: &str,