From f04e5ff612c70d086cb82d546d574a901d24fb56 Mon Sep 17 00:00:00 2001 From: Eric Wang Date: Thu, 22 Oct 2026 22:53:13 +0800 Subject: [PATCH] feat(services): per-session serialization for chat/stream/delete (A2 concurrency isolation) FastGPT serializes requests per chatId server-side; the LangGraph backend held all state in memory with no locking, so two concurrent requests on one session could interleave (checkpoint write races, garbled no_reply/turn bookkeeping). - AccidentAppService holds per-session asyncio locks; chat(), stream() and delete_session() execute under the session lock (LLM call included - same-session turns are strictly serialized, different sessions never contend) - set_info/get_info stay synchronous and lock-free: repository load/apply_update are atomic dict operations with no mixed intermediate state; a racing set vs chat yields one complete value - delete pops the lock entry after cleanup; a chat that arrived concurrently executes afterwards against the fresh (deleted) session - matching FastGPT's delete-then-chat new-conversation semantics - tests: same-session concurrent chats serialize (interval pairing probe), different sessions run parallel, delete waits for in-flight chat, chat-vs-set interleave leaves no torn values Verified: 219 unit tests (4 new concurrency tests) --- src/services/app_service.py | 45 ++++++++++--- test/services/test_app_service.py | 101 ++++++++++++++++++++++++++++++ 2 files changed, 139 insertions(+), 7 deletions(-) diff --git a/src/services/app_service.py b/src/services/app_service.py index 2f0f93b..7ccfd33 100644 --- a/src/services/app_service.py +++ b/src/services/app_service.py @@ -5,6 +5,11 @@ （对照旧 FastGPT 实现的“辅助对话 + 删记录”hack）； - chat 每轮从 repository 读最新业务字段注入图（set_info 写入后 下一轮 chat 必须可见），图产出字段差量后写回 repository； +- **同会话串行**（A2 并发隔离）：chat / stream / delete_session 按 + session_id 互斥执行——FastGPT 服务端对同一 chatId 串行处理， + 内存态无锁会交错（checkpoint 写竞争、簿记错乱）。不同会话 + 无竞争。set/get 保持同步无锁：repository 的 load/apply_update + 是 dict 原子操作，不存在混合中间态。 - 未知 key 兼容策略：get 对未知 key 返回 JSON 空串 '""'（现状兼容）； set 对未知 key 拒绝写入（白名单保护，差异见兼容清单）； - includeInputInfo 继续被 endpoint 接受，但本层不再使用 @@ -13,6 +18,8 @@ endpoint 只负责校验、协议（SSE / 响应模型）与依赖注入。 """ +import asyncio + from ..backends.chat import ChatBackend, ChatInput, ChatResult, FormUpdate from .fields_registry import ( FIELD_REGISTRY, @@ -35,6 +42,17 @@ class AccidentAppService: def __init__(self, backend: ChatBackend, repository: SessionRepository): self._backend = backend self._repository = repository + # 会话级互斥锁：同 session 的 chat/stream/delete 串行执行。 + # delete 成功后清理锁项；清理与等待者的竞争语义见 + # test_concurrent_delete_then_chat_behaves_as_fresh_session。 + self._locks: dict[str, asyncio.Lock] = {} + + def _lock_for(self, session_id: str) -> asyncio.Lock: + lock = self._locks.get(session_id) + if lock is None: + lock = asyncio.Lock() + self._locks[session_id] = lock + return lock # ---- chat：图每轮读取最新业务状态，产出差量写回 ---- @@ -69,6 +87,10 @@ class AccidentAppService: async def chat(self, chat_input: ChatInput) -> ChatResult: """非流式一轮对话：注入最新业务字段 -> 执行图 -> 写回字段差量。""" + async with self._lock_for(chat_input.session_id): + return await self._chat_locked(chat_input) + + async def _chat_locked(self, chat_input: ChatInput) -> ChatResult: injected = self._injected_fields(chat_input.session_id) result = await self._backend.complete( ChatInput( @@ -87,6 +109,11 @@ class AccidentAppService: async def stream(self, chat_input: ChatInput): """流式代理：字段注入同 chat()；FormUpdate 事件捕获写回后透传。""" + async with self._lock_for(chat_input.session_id): + async for event in self._stream_locked(chat_input): + yield event + + async def _stream_locked(self, chat_input: ChatInput): injected = self._injected_fields(chat_input.session_id) result_input = ChatInput( session_id=chat_input.session_id, @@ -151,15 +178,19 @@ class AccidentAppService: drop_checkpoint：由 core 层注入的 checkpoint 清理函数 （图生命周期归 lifespan 管理，本层不持有图引用）。 + 与会话锁互斥：删除不与在途 chat 交错。删除后清理锁项。 repository 删除幂等；checkpoint 缺失时静默（幂等语义一致）。 """ - self._repository.delete(session_id) - if drop_checkpoint is not None: - try: - await drop_checkpoint(session_id) - except Exception: - # checkpoint 已不存在视为删除成功的一部分（幂等） - pass + async with self._lock_for(session_id): + self._repository.delete(session_id) + if drop_checkpoint is not None: + try: + await drop_checkpoint(session_id) + except Exception: + # checkpoint 已不存在视为删除成功的一部分（幂等） + pass + # 清理锁项：引用随等待者自然释放，新会话重建新锁 + self._locks.pop(session_id, None) __all__ = ["AccidentAppService", "SET_INFO_INVALID_KEY_MSG"] diff --git a/test/services/test_app_service.py b/test/services/test_app_service.py index 81720ca..ae17de5 100644 --- a/test/services/test_app_service.py +++ b/test/services/test_app_service.py @@ -220,3 +220,104 @@ def _input(session_id, text): from src.backends.chat import ChatInput return ChatInput(session_id=session_id, text=text, client_mode="direct") + + +# ---- 并发隔离（A2）：同会话串行、不同会话并行、delete 与 chat 互斥 ---- + +import asyncio + + +class SerialProbeBackend: + """记录 complete 执行区间的 backend：start/end 事件对可判断是否交错。""" + + def __init__(self, delay: float = 0.03): + self.events: list[tuple[str, str]] = [] + self._delay = delay + + async def complete(self, chat_input): + self.events.append(("start", chat_input.text)) + await asyncio.sleep(self._delay) + self.events.append(("end", chat_input.text)) + return ChatResult(content="1002收到") + + async def stream(self, chat_input): + self.events.append(("start", chat_input.text)) + await asyncio.sleep(self._delay) + self.events.append(("end", chat_input.text)) + yield TextDelta("1002收到") + + +def _paired(events: list[tuple[str, str]]) -> bool: + """每个 start 与其 end 相邻配对（无交错）。顺序任意。""" + stack: list[str] = [] + for kind, tag in events: + if kind == "start": + if stack: + return False + stack.append(tag) + else: + if not stack or stack.pop() != tag: + return False + return not stack + + +@pytest.mark.asyncio +async def test_concurrent_chats_same_session_serialize(): + """同会话两个并发 chat 必须串行（区间不交错）。""" + backend = SerialProbeBackend() + service = AccidentAppService(backend, InMemorySessionRepository()) + await asyncio.gather( + service.chat(_input("s1", "a")), + service.chat(_input("s1", "b")), + ) + assert sorted(e[1] for e in backend.events) == ["a", "a", "b", "b"] + assert _paired(backend.events) + + +@pytest.mark.asyncio +async def test_concurrent_chats_different_sessions_run_parallel(): + """不同会话的 chat 互不阻塞（第二个 start 在第一个 end 之前）。""" + backend = SerialProbeBackend(delay=0.05) + service = AccidentAppService(backend, InMemorySessionRepository()) + await asyncio.gather( + service.chat(_input("s1", "a")), + service.chat(_input("s2", "b")), + ) + starts_before_first_end = [ + tag for kind, tag in backend.events[:2] if kind == "start" + ] + assert sorted(starts_before_first_end) == ["a", "b"] + + +@pytest.mark.asyncio +async def test_concurrent_delete_serialization(): + """在途 chat 与 delete 互斥：delete 等 chat 完成后才执行。""" + backend = SerialProbeBackend() + service = AccidentAppService(backend, InMemorySessionRepository()) + + async def noop_drop(_session_id): + backend.events.append(("drop", "s1")) + + await asyncio.gather( + service.chat(_input("s1", "a")), + service.delete_session("s1", drop_checkpoint=noop_drop), + ) + # drop 必须出现在 chat 的 end 之后 + kinds = [e for e in backend.events] + drop_index = next(i for i, e in enumerate(kinds) if e[0] == "drop") + end_index = next(i for i, e in enumerate(kinds) if e == ("end", "a")) + assert drop_index > end_index + + +@pytest.mark.asyncio +async def test_concurrent_chat_and_set_info_no_corruption(): + """set_info（同步无锁）与并发 chat 交错：repository 键值保持完整。""" + backend = SerialProbeBackend() + service = AccidentAppService(backend, InMemorySessionRepository()) + results = await asyncio.gather( + service.chat(_input("s1", "a")), + asyncio.to_thread(service.set_info, "s1", "hphm1", "苏B12345"), + ) + assert results[1] == ("200", "") + # 无论时序如何，最终两个写者之一的车牌值完整存在（无撕裂） + assert json.loads(service.get_info("s1", "hphm1")) in ("苏B12345", "")