From 7cb170e0c24f3a43d4cdd58726c02b799e6f39e8 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Wed, 12 Aug 2026 08:24:46 +0800 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20=E5=8E=8B=E7=BC=A9=E6=9C=9F?= =?UTF-8?q?=E9=97=B4=E9=87=8A=E6=94=BE=20session=20=E9=94=81=EF=BC=8C?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=96=B0=E5=BB=BA=E8=AF=9D=E9=A2=98=E5=88=97?= =?UTF-8?q?=E8=A1=A8=E5=88=B7=E6=96=B0=E5=BB=B6=E8=BF=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 压缩任务在 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 完全并行不受影响。 --- src/gateway/compaction.rs | 53 ++++++++++++++++++++++++++------------- 1 file changed, 35 insertions(+), 18 deletions(-) diff --git a/src/gateway/compaction.rs b/src/gateway/compaction.rs index 4444dce..15b4e5d 100644 --- a/src/gateway/compaction.rs +++ b/src/gateway/compaction.rs @@ -6,13 +6,20 @@ use crate::agent::AgentError; 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 -/// session lock during the LLM calls (2–5 seconds). Since the agent loop -/// has already finished by this point there is no response-time impact, and -/// the synchronous guarantee means the next execution always starts with -/// freshly compacted history. +/// The session lock is held only for the brief data-gathering and +/// history-reload phases. The expensive LLM call (2–5 seconds) and the DB +/// write happen **without** holding the session lock, so other commands +/// (e.g. `create_session`, `list_topics`) are not blocked during compression. +/// +/// 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。 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 topic_id = topic_id.into(); - let mut session_guard = session.lock().await; - session_guard.ensure_persistent_session(&chat_id)?; - session_guard.ensure_chat_loaded(&chat_id, Some(&topic_id))?; + // Phase 1: brief session lock to gather compaction inputs. + let (history, compressor, store, session_id, provider_config) = { + 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 compressor = session_guard.compressor().clone(); + let history = session_guard.get_or_create_history(&topic_id).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) { 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!( chat_id = %chat_id, topic_id = %topic_id, 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, // so there is no risk of orphaned tool call sequences. let compressed = compressor .compress_two_segment(&history, &provider_config) .await?; + // Phase 3: DB write — store is Arc, no + // session lock needed. // 保留原始消息（标记 is_compacted=1）+ 插入压缩摘要，不删除原消息， // 从而让前端仍能展示完整原始对话，LLM 只看压缩后的精简历史。 store @@ -65,7 +78,11 @@ pub(crate) async fn schedule_background_history_compaction( "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(()) }