fix(gateway): 压缩期间释放 session 锁,修复新建话题列表刷新延迟

压缩任务在 LLM 调用(2-5s)期间持有 session 锁,阻塞 create_session
命令获取锁执行 switch_topic,导致新建话题后列表刷新延迟。

将压缩重构为 4 阶段:
1. 短暂持锁:读取 history/compressor/store/session_id/provider_config
2. 释放锁:LLM 压缩调用(2-5s)
3. 释放锁:DB 写入(store 为 Arc,DB 层自带事务保护)
4. 短暂持锁:reload 内存历史

并发安全:同 topic 由调用方的 per-topic serial lock 保证串行;
不同 topic 完全并行不受影响。
This commit is contained in:
oudecheng 2026-08-12 08:24:46 +08:00
parent 77a0eac2c8
commit 7cb170e0c2

View File

@ -6,13 +6,20 @@ use crate::agent::AgentError;
use super::session::Session; use super::session::Session;
/// Run two-segment history compression synchronously. /// Run two-segment history compression.
/// ///
/// Unlike the previous background approach (tokio::spawn), this holds the /// The session lock is held only for the brief data-gathering and
/// session lock during the LLM calls (25 seconds). Since the agent loop /// history-reload phases. The expensive LLM call (25 seconds) and the DB
/// has already finished by this point there is no response-time impact, and /// write happen **without** holding the session lock, so other commands
/// the synchronous guarantee means the next execution always starts with /// (e.g. `create_session`, `list_topics`) are not blocked during compression.
/// freshly compacted history. ///
/// Concurrency safety:
/// - **Same topic**: the caller (`prepare_and_execute_message`) holds the
/// per-topic serial lock (`_serial_guard`) for the entire duration of
/// execution + compaction, so no other message for this topic can modify
/// the in-memory or DB history between phases.
/// - **Different topic**: fully unblocked — the session lock is free during
/// the LLM call.
/// ///
/// 按 topic_id 隔离:压缩只处理指定 topic 的历史DB 替换也只影响该 topic。 /// 按 topic_id 隔离:压缩只处理指定 topic 的历史DB 替换也只影响该 topic。
pub(crate) async fn schedule_background_history_compaction( pub(crate) async fn schedule_background_history_compaction(
@ -23,35 +30,41 @@ pub(crate) async fn schedule_background_history_compaction(
let chat_id = chat_id.into(); let chat_id = chat_id.into();
let topic_id = topic_id.into(); let topic_id = topic_id.into();
let mut session_guard = session.lock().await; // Phase 1: brief session lock to gather compaction inputs.
session_guard.ensure_persistent_session(&chat_id)?; let (history, compressor, store, session_id, provider_config) = {
session_guard.ensure_chat_loaded(&chat_id, Some(&topic_id))?; let mut session_guard = session.lock().await;
session_guard.ensure_persistent_session(&chat_id)?;
session_guard.ensure_chat_loaded(&chat_id, Some(&topic_id))?;
let history = session_guard.get_or_create_history(&topic_id).clone(); let history = session_guard.get_or_create_history(&topic_id).clone();
let compressor = session_guard.compressor().clone(); let compressor = session_guard.compressor().clone();
let store = session_guard.store();
let session_id = session_guard.persistent_session_id(&chat_id);
let provider_config = session_guard.provider_config().clone();
(history, compressor, store, session_id, provider_config)
};
// session lock released here
if !compressor.should_compress(&history) { if !compressor.should_compress(&history) {
return Ok(()); return Ok(());
} }
let store = session_guard.store();
let session_id = session_guard.persistent_session_id(&chat_id);
let provider_config = session_guard.provider_config().clone();
tracing::info!( tracing::info!(
chat_id = %chat_id, chat_id = %chat_id,
topic_id = %topic_id, topic_id = %topic_id,
msg_count = history.len(), msg_count = history.len(),
"Starting synchronous two-segment compression" "Starting two-segment compression (session lock released during LLM call)"
); );
// Synchronous compression — holds lock during LLM calls. // Phase 2: LLM compression WITHOUT holding the session lock.
// compress_two_segment guarantees the result contains no tool_calls, // compress_two_segment guarantees the result contains no tool_calls,
// so there is no risk of orphaned tool call sequences. // so there is no risk of orphaned tool call sequences.
let compressed = compressor let compressed = compressor
.compress_two_segment(&history, &provider_config) .compress_two_segment(&history, &provider_config)
.await?; .await?;
// Phase 3: DB write — store is Arc<dyn ConversationRepository>, no
// session lock needed.
// 保留原始消息(标记 is_compacted=1+ 插入压缩摘要,不删除原消息, // 保留原始消息(标记 is_compacted=1+ 插入压缩摘要,不删除原消息,
// 从而让前端仍能展示完整原始对话LLM 只看压缩后的精简历史。 // 从而让前端仍能展示完整原始对话LLM 只看压缩后的精简历史。
store store
@ -65,7 +78,11 @@ pub(crate) async fn schedule_background_history_compaction(
"Two-segment compression committed (original messages retained)" "Two-segment compression committed (original messages retained)"
); );
session_guard.reload_topic_history(&chat_id, &topic_id)?; // Phase 4: re-acquire session lock to refresh in-memory history.
{
let mut session_guard = session.lock().await;
session_guard.reload_topic_history(&chat_id, &topic_id)?;
}
Ok(()) Ok(())
} }