perf(db): 热路径同步 SQLite 操作改 spawn_blocking 并定向查询
- processor/session 热点 DB 调用放到 blocking 线程池，避免阻塞 tokio worker - 定向查询 topic 首条用户消息（DB 侧 LIMIT 1），不再全量加载历史 - 新增 first_user_message_content 访问方法
This commit is contained in:
oudecheng
parent
f2fc5e97ac
commit
4517e4a724
@ -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,
|
||||
|
||||
@ -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<SessionStore>,
|
||||
channel_name: &str,
|
||||
chat_id: &str,
|
||||
metadata: &HashMap<String, String>,
|
||||
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"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -1785,6 +1785,33 @@ impl SessionStore {
|
||||
}
|
||||
}
|
||||
|
||||
/// 定向查询指定话题的第一条 user 消息内容。
|
||||
///
|
||||
/// 数据库侧 `LIMIT 1`，避免为取单条消息全量加载并反序列化整个话题历史
|
||||
/// （话题越长，全量加载的 CPU/内存浪费越大）。
|
||||
pub fn first_user_message_content(
|
||||
&self,
|
||||
topic_id: &str,
|
||||
) -> Result<Option<String>, 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(*)` 在数据库侧计数，避免将所有消息
|
||||
|
||||
@ -36,6 +36,10 @@ pub trait ConversationRepository: Send + Sync + 'static {
|
||||
session_id: Option<&str>,
|
||||
) -> Result<Vec<ChatMessage>, StorageError>;
|
||||
|
||||
/// 定向查询指定话题的第一条 user 消息内容（数据库侧 LIMIT 1）。
|
||||
/// 避免为取单条消息而全量加载并反序列化整个话题历史。
|
||||
fn first_user_message_content(&self, topic_id: &str) -> Result<Option<String>, 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<Option<String>, StorageError> {
|
||||
super::SessionStore::first_user_message_content(self, topic_id)
|
||||
}
|
||||
|
||||
fn compact_topic_history(
|
||||
&self,
|
||||
session_id: &str,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user