fix(langgraph): 请求日志中间件重构为纯ASGI - 避免BaseHTTPMiddleware消费body导致POST死锁

cfc-langgraph/app/main.py

@@ -3,18 +3,17 @@ import time
 import asyncio
 import logging
 from fastapi import FastAPI, Request
-from starlette.middleware.base import BaseHTTPMiddleware
 from app.api import health, recommend, chat, analyze, tongue, adapter, report_parse, meal, logs, audio
 from app import monitoring
 from src.app import router as questionnaire_router
-from app.middleware import request_log_middleware
+from app.middleware import RequestLogMiddleware
 
 logger = logging.getLogger(__name__)
 
 app = FastAPI(title="cfc-langgraph", version="0.3.0")
 
-# 注册请求日志中间件（必须在其他路由之前）
-app.add_middleware(BaseHTTPMiddleware, dispatch=request_log_middleware)
+# 注册请求日志中间件（纯 ASGI，避免 BaseHTTPMiddleware 消费 body 导致单 worker 死锁）
+app.add_middleware(RequestLogMiddleware)
 
 app.include_router(health.router)
 app.include_router(recommend.router)

cfc-langgraph/app/middleware.py

@@ -1,9 +1,14 @@
-"""请求日志中间件 — 记录最近 MAX_ENTRIES 条请求到内存"""
+"""请求日志中间件 — 记录最近 MAX_ENTRIES 条请求到内存
+
+实现为纯 ASGI 中间件（而非 BaseHTTPMiddleware），原因：
+- BaseHTTPMiddleware 中先 `await request.body()` 再 `call_next(request)` 会消费请求体，
+  导致下游 handler 再次 receive 时拿不到 body，POST 请求死锁/挂起；
+- 单 worker 部署下，一个请求挂起会卡死整个 worker，所有请求（含 /api/v1/logs）超时。
+
+纯 ASGI 中间件通过包装 receive/send 通道缓存请求体与响应状态，既记录日志又不破坏下游请求。
+"""
 import time
-import asyncio
 from collections import deque
-from typing import Optional
-from fastapi import Request, Response
 import logging
 
 logger = logging.getLogger(__name__)
@@ -24,76 +29,99 @@ def clear_request_log() -> int:
     return n
 
 
-async def request_log_middleware(request: Request, call_next) -> Response:
-    start_time = time.perf_counter()
-    client_ip = request.client.host if request.client else "-"
-    
-    # 读取请求体（不重复消费）
-    body = None
-    if request.method in ("POST", "PUT", "PATCH"):
-        try:
-            raw = await request.body()
-            if raw:
-                try:
-                    body = raw.decode("utf-8")
-                    if len(body) > 2000:
-                        body = body[:2000] + "...(truncated)"
-                except Exception:
-                    body = f"<binary {len(raw)} bytes>"
-        except Exception:
-            pass
-    
+def _decode_body(raw: bytes) -> str:
+    """将请求体字节解码为可读文本（截断超长内容）"""
+    if not raw:
+        return ""
     try:
-        response = await call_next(request)
-    except Exception as exc:
+        text = raw.decode("utf-8")
+        if len(text) > 2000:
+            text = text[:2000] + "...(truncated)"
+        return text
+    except Exception:
+        return f"<binary {len(raw)} bytes>"
+
+
+class RequestLogMiddleware:
+    """纯 ASGI 请求日志中间件。
+
+    通过包装 receive 通道缓存请求体（不消费、不阻塞下游），
+    通过包装 send 通道捕获响应状态码与响应体，避免破坏原请求。
+    """
+
+    def __init__(self, app):
+        self.app = app
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        start_time = time.perf_counter()
+        method = scope.get("method", "GET")
+        path = scope.get("path", "")
+        client = scope.get("client")
+        client_ip = client[0] if client else "-"
+
+        # 缓存请求体（仅对带 body 的方法），不影响下游 receive
+        body_bytes = b""
+        body_cached = False
+
+        async def receive_wrapper():
+            nonlocal body_bytes, body_cached
+            message = await receive()
+            if message["type"] == "http.request":
+                body_bytes += message.get("body", b"")
+                if not message.get("more_body", False):
+                    body_cached = True
+            return message
+
+        # 捕获响应状态码与响应体
+        status_code = None
+        response_chunks = []
+        sent_response_start = False
+
+        async def send_wrapper(message):
+            nonlocal status_code, response_chunks, sent_response_start
+            if message["type"] == "http.response.start":
+                status_code = message.get("status", 500)
+                sent_response_start = True
+            elif message["type"] == "http.response.body":
+                body = message.get("body", b"")
+                if body:
+                    response_chunks.append(body)
+                    total = sum(len(c) for c in response_chunks)
+                    if total > 1000:
+                        response_chunks[:] = [b"".join(response_chunks)[:1000]]
+            return await send(message)
+
+        try:
+            await self.app(scope, receive_wrapper, send_wrapper)
+        except Exception as exc:
+            logger.error("REQUEST EXCEPTION: %s %s %.2fs error=%r",
+                         method, path, time.perf_counter() - start_time, exc)
+            if not sent_response_start:
+                from starlette.responses import JSONResponse
+                resp = JSONResponse(status_code=500, content={"detail": str(exc)})
+                await resp(scope, receive, send)
+            status_code = status_code or 500
+
         elapsed = time.perf_counter() - start_time
-        status_code = 500
-        response_body = ""
-        logger.error("REQUEST EXCEPTION: %s %s %.2fs error=%r",
-                     request.method, request.url.path, elapsed, exc)
-        _request_log.appendleft({
+        final_status = status_code if status_code is not None else 500
+
+        entry = {
             "ts": time.strftime("%H:%M:%S"),
-            "method": request.method,
-            "path": request.url.path,
-            "status": 500,
+            "method": method,
+            "path": path,
+            "status": final_status,
             "duration_ms": round(elapsed * 1000),
             "client_ip": client_ip,
-            "body": body,
-            "error": str(exc)[:500],
-        })
-        # 构造错误响应
-        from fastapi.responses import JSONResponse
-        return JSONResponse(status_code=500, content={"detail": str(exc)})
-    
-    elapsed = time.perf_counter() - start_time
-    status_code = response.status_code
-    
-    # 读取响应体（不可逆，只取内容长度）
-    response_body = ""
-    if hasattr(response, "body") and response.body:
-        try:
-            response_body = response.body.decode("utf-8")[:1000]
-        except Exception:
-            response_body = f"<binary {len(response.body)} bytes>"
-    
-    # 写入日志
-    entry = {
-        "ts": time.strftime("%H:%M:%S"),
-        "method": request.method,
-        "path": request.url.path,
-        "status": status_code,
-        "duration_ms": round(elapsed * 1000),
-        "client_ip": client_ip,
-        "body": body,
-        "response": response_body,
-    }
-    _request_log.appendleft(entry)
-    
-    # 超5秒标记 warning
-    if elapsed > 5:
-        logger.warning("SLOW_REQUEST: %s %s %.2fs", request.method, request.url.path, elapsed)
-    else:
-        logger.debug("REQUEST: %s %s %.2fs", request.method, request.url.path, elapsed)
-    
-    response.headers["X-Response-Time"] = f"{elapsed:.3f}s"
-    return response
+            "body": _decode_body(body_bytes) if body_cached else None,
+            "response": _decode_body(b"".join(response_chunks)[:1000]) if response_chunks else "",
+        }
+        _request_log.appendleft(entry)
+
+        if elapsed > 5:
+            logger.warning("SLOW_REQUEST: %s %s %.2fs", method, path, elapsed)
+        else:
+            logger.debug("REQUEST: %s %s %.2fs", method, path, elapsed)