diff --git a/src/gateway/execution.rs b/src/gateway/execution.rs index 5713018..2e1b58b 100644 --- a/src/gateway/execution.rs +++ b/src/gateway/execution.rs @@ -200,15 +200,16 @@ impl AgentExecutionService { &self, request: MessageExecutionRequest<'_>, ) -> Result, AgentError> { - // 获取该 chat 的串行锁(通过短暂获取 session 锁) - // 同一 chat 的消息处理必须串行执行,防止并发 loop 操作同一历史的不同快照 + // 获取该 topic 的串行锁(通过短暂获取 session 锁) + // 同一 topic 的消息处理必须串行执行,防止并发 loop 操作同一历史的不同快照 + // 不同 topic 之间互不阻塞,支持多话题并发执行 let serial_lock = { let mut session_guard = request.session.lock().await; - session_guard.chat_serial_lock(request.chat_id) + session_guard.topic_serial_lock(request.chat_id) }; - // 等待该 chat 的前一条消息处理完成(含压缩) - // await 串行锁时不持有 session 锁,其他 chat 的消息可以正常处理 + // 等待该 topic 的前一条消息处理完成(含压缩) + // await 串行锁时不持有 session 锁,其他 topic 的消息可以正常处理 let _serial_guard = serial_lock.lock().await; let (history, agent, user_message, user_message_count, original_topic_id) = { @@ -291,13 +292,13 @@ impl AgentExecutionService { &self, request: ScheduledExecutionRequest<'_>, ) -> Result, AgentError> { - // 获取该 chat 的串行锁(与普通消息路径共享,保证串行执行) + // 获取该 topic 的串行锁(与普通消息路径共享,保证串行执行) let serial_lock = { let mut session_guard = request.session.lock().await; - session_guard.chat_serial_lock(request.chat_id) + session_guard.topic_serial_lock(request.chat_id) }; - // 等待该 chat 的前一条消息处理完成(含压缩) + // 等待该 topic 的前一条消息处理完成(含压缩) let _serial_guard = serial_lock.lock().await; let (history, mut agent, user_message, user_message_count, original_topic_id, store, session_id) = { @@ -500,9 +501,9 @@ mod tests { assert!(should_display_message_to_user(true, &message)); } - /// 对抗性测试:同一 chat 的串行锁被持有时,第二次获取应阻塞 + /// 对抗性测试:同一 topic 的串行锁被持有时,第二次获取应阻塞 #[tokio::test] - async fn test_chat_serial_lock_blocks_concurrent_access() { + async fn test_topic_serial_lock_blocks_concurrent_access() { let lock = std::sync::Arc::new(tokio::sync::Mutex::new(())); let _guard1 = lock.lock().await; @@ -516,9 +517,9 @@ mod tests { assert!(result.is_err(), "第二次获取同一锁应阻塞"); } - /// 对抗性测试:不同 chat 的串行锁互不影响,可同时获取 + /// 对抗性测试:不同 topic 的串行锁互不影响,可同时获取 #[tokio::test] - async fn test_different_chat_locks_independent() { + async fn test_different_topic_locks_independent() { let lock_a = std::sync::Arc::new(tokio::sync::Mutex::new(())); let lock_b = std::sync::Arc::new(tokio::sync::Mutex::new(())); @@ -531,7 +532,7 @@ mod tests { ) .await; - assert!(result.is_ok(), "不同 chat 的锁应互不影响"); + assert!(result.is_ok(), "不同 topic 的锁应互不影响"); } /// 对抗性测试:错误返回路径锁被正确释放(RAII 保证) diff --git a/src/gateway/session.rs b/src/gateway/session.rs index fee3ed3..b73afc4 100644 --- a/src/gateway/session.rs +++ b/src/gateway/session.rs @@ -533,10 +533,11 @@ impl Session { &self.compressor } - /// 获取该 chat 的串行化锁。 - /// 同一 chat 的消息处理(agent loop + 压缩)共享此锁,保证串行执行。 - pub(crate) fn chat_serial_lock(&mut self, chat_id: &str) -> Arc> { - self.history.chat_serial_lock(chat_id) + /// 获取该 topic 的串行化锁。 + /// 同一 topic 的消息处理(agent loop + 压缩)共享此锁,保证串行执行; + /// 不同 topic 之间互不阻塞。 + pub(crate) fn topic_serial_lock(&mut self, topic_id: &str) -> Arc> { + self.history.topic_serial_lock(topic_id) } pub(crate) fn reload_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> { diff --git a/src/gateway/session_history.rs b/src/gateway/session_history.rs index 685e698..8985f85 100644 --- a/src/gateway/session_history.rs +++ b/src/gateway/session_history.rs @@ -21,10 +21,11 @@ pub(crate) struct SessionHistory { chat_topic_ids: HashMap, // 每个 chat 的当前 topic history_topic_ids: HashMap, // 每个 chat 的历史所对应的话题 compression_in_flight: HashSet, - /// 按 chat_id 的串行化锁。 - /// 同一 chat 的消息处理(agent loop + 压缩)必须串行执行, + /// 按 topic_id 的串行化锁。 + /// 同一 topic 的消息处理(agent loop + 压缩)必须串行执行, /// 防止并发 loop 操作同一历史的不同快照产生交错序列。 - chat_serial_locks: HashMap>>, + /// 不同 topic 之间互不阻塞,支持多话题并发执行。 + topic_serial_locks: HashMap>>, conversations: Arc, skill_events: Arc, } @@ -41,17 +42,18 @@ impl SessionHistory { chat_topic_ids: HashMap::new(), history_topic_ids: HashMap::new(), compression_in_flight: HashSet::new(), - chat_serial_locks: HashMap::new(), + topic_serial_locks: HashMap::new(), conversations, skill_events, } } - /// 获取或创建该 chat 的串行化锁。 - /// 同一 chat 的所有消息处理共享同一个锁,保证串行执行。 - pub(crate) fn chat_serial_lock(&mut self, chat_id: &str) -> Arc> { - self.chat_serial_locks - .entry(chat_id.to_string()) + /// 获取或创建该 topic 的串行化锁。 + /// 同一 topic 的所有消息处理共享同一个锁,保证串行执行; + /// 不同 topic 之间互不阻塞,支持多话题并发执行。 + pub(crate) fn topic_serial_lock(&mut self, topic_id: &str) -> Arc> { + self.topic_serial_locks + .entry(topic_id.to_string()) .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(()))) .clone() }