scope request dedupe to explicit request ids

Author: LJR199887

README.md

@@ -251,6 +251,7 @@ curl -X POST "http://127.0.0.1:6001/v1/images/generations" \
 - 可用于：`/v1/chat/completions` 和 `/v1/images/generations`
 - 查询接口：`GET /v1/requests/{request_id}`
 - 服务会在响应头回写 `X-Request-Id`，同时在 JSON 响应体里也包含 `request_id`
+- 同一个 `request_id` / `X-Request-Id` 会被视为同一任务的幂等键；如果换了真实请求内容，请务必使用新的 `request_id`
 
 提交示例：
 

app.py

@@ -251,7 +251,7 @@ def _normalize_request_id(value: Any) -> str:
     return text
 
 
-def _build_request_dedupe_key(request: Request, path: str, raw_body: bytes) -> str:
+def _build_request_payload_hash(raw_body: bytes) -> str:
     if not raw_body:
         return ""
     try:
@@ -262,36 +262,23 @@ def _build_request_dedupe_key(request: Request, path: str, raw_body: bytes) -> s
         return ""
     if bool(data.get("stream")):
         return ""
-
     normalized_data = dict(data)
-    explicit_request_id = _normalize_request_id(
-        normalized_data.pop("request_id", None)
-        or normalized_data.pop("requestId", None)
-    )
+    normalized_data.pop("request_id", None)
+    normalized_data.pop("requestId", None)
     normalized_body = json.dumps(
         normalized_data,
         ensure_ascii=False,
         sort_keys=True,
         separators=(",", ":"),
     )
-    client_host = str(
-        getattr(getattr(request, "client", None), "host", "") or ""
-    ).strip()
-    user_agent = str(request.headers.get("user-agent") or "").strip()
-    auth_value = str(
-        request.headers.get("authorization")
-        or request.headers.get("x-api-key")
-        or ""
-    ).strip()
-    scope_seed = "|".join((client_host, user_agent, auth_value))
-    scope_hash = hashlib.sha256(scope_seed.encode("utf-8")).hexdigest()[:16]
-    dedupe_seed = (
-        f"rid:{explicit_request_id}|{normalized_body}"
-        if explicit_request_id
-        else normalized_body
-    )
-    dedupe_hash = hashlib.sha256(dedupe_seed.encode("utf-8")).hexdigest()
-    return f"{path}|{scope_hash}|{dedupe_hash}"
+    return hashlib.sha256(normalized_body.encode("utf-8")).hexdigest()
+
+
+def _build_request_dedupe_key(path: str, explicit_request_id: str) -> str:
+    request_id = _normalize_request_id(explicit_request_id)
+    if not request_id:
+        return ""
+    return f"{path}|rid|{request_id}"
 
 
 def _prune_request_dedupe_entries(now_ts: Optional[float] = None) -> None:
@@ -311,26 +298,30 @@ def _prune_request_dedupe_entries(now_ts: Optional[float] = None) -> None:
 
 
 def _begin_request_dedupe(
-    dedupe_key: str, request_id: str
-) -> tuple[bool, dict[str, Any] | None]:
+    dedupe_key: str, request_id: str, payload_hash: str
+) -> tuple[str, dict[str, Any] | None]:
     if not dedupe_key:
-        return True, None
+        return "owner", None
     now_ts = time.time()
     with _request_dedupe_lock:
         _prune_request_dedupe_entries(now_ts)
         existing = _request_dedupe_entries.get(dedupe_key)
         if isinstance(existing, dict):
+            existing_payload_hash = str(existing.get("payload_hash") or "")
+            if existing_payload_hash and payload_hash and existing_payload_hash != payload_hash:
+                return "conflict", existing
             existing["updated_at"] = now_ts
-            return False, existing
+            return "join", existing
         entry = {
             "request_id": request_id,
+            "payload_hash": payload_hash,
             "created_at": now_ts,
             "updated_at": now_ts,
             "event": threading.Event(),
             "response": None,
         }
         _request_dedupe_entries[dedupe_key] = entry
-        return True, entry
+        return "owner", entry
 
 
 def _freeze_response(response: Response) -> dict[str, Any]:
@@ -718,23 +709,43 @@ async def request_logger(request: Request, call_next):
             header_request_id = _normalize_request_id(
                 request.headers.get("x-request-id")
             )
+            explicit_request_id = (
+                header_request_id or str(body_meta.get("request_id") or "").strip()
+            )
             request.state.log_id = (
-                header_request_id
-                or str(body_meta.get("request_id") or "").strip()
+                explicit_request_id
                 or uuid.uuid4().hex[:12]
             )
-            dedupe_key = _build_request_dedupe_key(request, path, raw_body)
-            dedupe_owner, dedupe_entry = _begin_request_dedupe(
+            dedupe_key = _build_request_dedupe_key(path, explicit_request_id)
+            dedupe_action, dedupe_entry = _begin_request_dedupe(
                 dedupe_key,
                 str(getattr(request.state, "log_id", "") or ""),
+                _build_request_payload_hash(raw_body),
             )
+            dedupe_owner = dedupe_action == "owner"
             if isinstance(dedupe_entry, dict):
                 request.state.log_id = str(
                     dedupe_entry.get("request_id")
                     or getattr(request.state, "log_id", "")
                     or uuid.uuid4().hex[:12]
                 )
-            if dedupe_key and not dedupe_owner and isinstance(dedupe_entry, dict):
+            if dedupe_action == "conflict":
+                conflict_payload = {
+                    "error": {
+                        "message": "request_id already exists with a different payload",
+                        "type": "invalid_request_error",
+                    }
+                }
+                response = Response(
+                    content=json.dumps(conflict_payload, ensure_ascii=False).encode("utf-8"),
+                    status_code=409,
+                    media_type="application/json",
+                )
+                response.headers["X-Request-Id"] = str(
+                    getattr(request.state, "log_id", "") or ""
+                )
+                return response
+            if dedupe_key and dedupe_action == "join" and isinstance(dedupe_entry, dict):
                 replay = await _replay_deduped_response(dedupe_entry)
                 if replay is not None:
                     dedupe_replayed = True