Skip to content

Commit d3be873

Browse files
committed
fix(gateway): AI网关中低危修复——熔断器语义/探活超时/JWKS安全/Key冷却/重试策略/错误脱敏/Redis重连/限流原子化
- GW-M1: 熔断器 CLOSED 成功重置失败计数(分散失败不再误熔断);仅 HALF_OPEN 允许恢复(冷却期不可绕过) - GW-M2: 健康探活独立 10s 超时(原最长 900s)+ 冷却期内探活成功不恢复 - GW-M3: JWKS 共享 AsyncClient 连接池 + asyncio.Lock 防击穿 + 网络失败 fail-closed(不再返回过期缓存) - GW-M4: KeyPool 全冷却返回 None(不再打必失败的冷却 Key);429 时自动标记当前 Key 冷却(熔断联动接线) - GW-M5: with_retry_and_timeout 仅重试可重试错误,401/400/429/内容审核直接上抛(消除 9 连发重试风暴) - GW-M6: 上游错误脱敏——响应仅返回通用文案与结构化字段,不再透传 provider 内部信息 - GW-M7: Redis 惰性重连——连接失效后下次操作自动恢复,限流/缓存不再静默失效至重启 - GW-M8: chunked 编码请求体流式计数限流(无 Content-Length 时逐块累计超限 413) - GW-M9: Prompt 防护移除 >1000 字符跳过 + 数组全量检测(长文注入不再绕过) - GW-M10: AI 响应缓存键前缀 user_id,杜绝跨用户隐私内容交叉泄漏 - GW-M11: 匿名用户按 IP 分桶限流,防止共享配额被污染 - GW-M12: 限流 Lua 脚本原子化(INCR+超限自动回滚),消除计数竞态与负数 - GW-M13: balance 共享 HTTP 客户端 + asyncio.Lock 防缓存击穿 - GW-M14: Key 轮换加 asyncio.Lock + 旧客户端 aclose 释放连接池 - GW-M15: 转写请求 base64 max_length + 参数范围校验;ASR confidence 不再硬编码 0.9(无数据源时返回 0.0) - GW-L1: 删除无调用者的 semantic_cache 死代码 - GW-L2: text_dedup 长度预过滤降低 O(n²) 实际开销 - GW-L3: _FEATURE_CONTEXT 成功路径 try/finally 重置 - GW-L5: 虚拟 Key 随机熵 64 位 → 128 位 测试: python -m pytest tests/ -q → 193 passed
1 parent 8e78b6f commit d3be873

19 files changed

Lines changed: 297 additions & 285 deletions

‎server/ai-gateway/cache/redis_cache.py‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -62,12 +62,16 @@ async def disconnect(self) -> None:
6262

6363
async def get(self, key: str) -> str | None:
6464
"""获取缓存值,Redis 不可用时返回 None"""
65+
if not self._client:
66+
await self._maybe_reconnect() # GW-M7: 惰性重连
6567
if not self._client:
6668
return None
6769
try:
6870
return await self._client.get(key)
6971
except Exception as exc:
7072
logger.warning("Redis GET 失败 key=%s: %s", key, exc)
73+
# 连接可能已失效,尝试重连供下次使用
74+
await self._maybe_reconnect()
7175
return None
7276

7377
async def set(self, key: str, value: str, expire: int = 0) -> bool:
@@ -76,6 +80,8 @@ async def set(self, key: str, value: str, expire: int = 0) -> bool:
7680
7781
Redis 不可用时静默跳过。返回是否设置成功。
7882
"""
83+
if not self._client:
84+
await self._maybe_reconnect() # GW-M7: 惰性重连
7985
if not self._client:
8086
return False
8187
try:
@@ -86,6 +92,7 @@ async def set(self, key: str, value: str, expire: int = 0) -> bool:
8692
return True
8793
except Exception as exc:
8894
logger.warning("Redis SET 失败 key=%s: %s", key, exc)
95+
await self._maybe_reconnect()
8996
return False
9097

9198
async def increment(self, key: str, expire: int = 86400) -> int:
@@ -102,6 +109,8 @@ async def increment(self, key: str, expire: int = 86400) -> int:
102109
Returns:
103110
int: 递增后的值;Redis 不可用时返回 0
104111
"""
112+
if not self._client:
113+
await self._maybe_reconnect() # GW-M7: 惰性重连
105114
if not self._client:
106115
return 0
107116
try:
@@ -112,8 +121,20 @@ async def increment(self, key: str, expire: int = 86400) -> int:
112121
return results[0]
113122
except Exception as exc:
114123
logger.warning("Redis INCR 失败 key=%s: %s", key, exc)
124+
await self._maybe_reconnect()
115125
return 0
116126

127+
async def _maybe_reconnect(self) -> None:
128+
"""
129+
GW-M7: 惰性重连——Redis 重启/网络抖动后连接失效,
130+
下次操作时自动尝试恢复,避免限流/缓存静默失效至进程重启。
131+
connect() 内部已捕获异常,此处不会向上抛出。
132+
"""
133+
try:
134+
await self.connect()
135+
except Exception:
136+
pass
137+
117138
async def get_ai_cache(self, prompt_hash: str) -> dict[str, Any] | None:
118139
"""
119140
获取 AI 响应缓存

‎server/ai-gateway/cache/semantic_cache.py‎

Lines changed: 0 additions & 195 deletions
This file was deleted.

‎server/ai-gateway/config/fallback.py‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -197,7 +197,12 @@ async def _run_fallback_chain():
197197
try:
198198
async def _do_call():
199199
_FEATURE_CONTEXT.set(feature)
200-
return await fn(provider, model_name)
200+
try:
201+
return await fn(provider, model_name)
202+
finally:
203+
# GW-L3: 成功路径也重置 context——避免同任务后续
204+
# 无 _feature 的 provider 调用继承上次 feature
205+
_FEATURE_CONTEXT.set("")
201206

202207
# Phase3: 并发信号量保护(视频/多模态走 heavy 上限)
203208
result = await _run_under_semaphore(app, feature, _do_call)

‎server/ai-gateway/config/key_pool.py‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ async def next_key(self) -> str | None:
6060
"""轮询获取下一个可用 Key
6161
6262
Returns:
63-
可用的 API Key,全部不可用时返回 None
63+
可用的 API Key,全部不可用时返回 None(调用方据此跳过该 Provider)
6464
"""
6565
if not self._keys:
6666
return None
@@ -73,12 +73,13 @@ async def next_key(self) -> str | None:
7373
self._index += 1
7474
if self._disabled_until.get(key, 0) <= now:
7575
return key
76-
# 全部不可用,返回第一个(让上层决定是否重试)
76+
# GW-M4: 全部不可用时返回 None 而非冷却中的 Key——
77+
# 返回冷却 Key 会让调用方对必失败的 Key 发起请求(429 重试风暴)
7778
logger.warning(
7879
"KeyPool [%s]: 所有 %d 个 Key 均处于冷却期",
7980
self.provider, len(self._keys),
8081
)
81-
return self._keys[0]
82+
return None
8283

8384
def mark_unavailable(self, key: str, cooldown: int = _KEY_COOLDOWN_SECONDS) -> None:
8485
"""标记 Key 暂时不可用(供应商返回 429/401 时调用)"""

‎server/ai-gateway/main.py‎

Lines changed: 17 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -214,15 +214,30 @@ async def ai_error_handler(request: Request, exc: AIError) -> JSONResponse:
214214
"AIError: %s (status=%d, path=%s)",
215215
exc.message, exc.status_code, request.url.path,
216216
)
217+
# GW-M6: 不向客户端透传上游错误细节——exc.message/reason 可能含
218+
# provider 内部信息、prompt 片段、配额详情,仅返回通用文案与结构化字段
219+
safe_detail = {
220+
k: v for k, v in exc.detail.items()
221+
if k in ("provider", "feature", "limit", "model")
222+
}
217223
return JSONResponse(
218224
status_code=exc.status_code,
219225
content={
220-
"detail": exc.message,
221-
**exc.detail,
226+
"detail": _GENERIC_ERROR_MESSAGES.get(exc.status_code, "AI 服务暂时不可用"),
227+
**safe_detail,
222228
},
223229
)
224230

225231

232+
# 上游错误脱敏后的通用错误文案(GW-M6)
233+
_GENERIC_ERROR_MESSAGES: dict[int, str] = {
234+
401: "认证失败,请重新登录",
235+
429: "请求过于频繁,请稍后重试",
236+
502: "模型服务响应异常,请稍后重试",
237+
503: "AI 服务暂时不可用,请稍后重试",
238+
}
239+
240+
226241
# ============================================================
227242
# 注册路由
228243
# ============================================================

‎server/ai-gateway/middleware/auth.py‎

Lines changed: 28 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,17 @@ def _normalize_pem_key(raw: str) -> str:
119119
_jwks_cache: Optional[dict] = None
120120
_jwks_cache_time: float = 0.0
121121
_JWKS_CACHE_TTL: int = 3600 # 缓存有效期(秒),默认 1 小时
122+
# GW-M3: 防缓存击穿锁 + 共享连接池客户端(避免每次请求新建 TCP/TLS 连接)
123+
_jwks_lock: asyncio.Lock = asyncio.Lock()
124+
_jwks_client: Optional[httpx.AsyncClient] = None
125+
126+
127+
def _get_jwks_client() -> httpx.AsyncClient:
128+
"""获取共享 JWKS HTTP 客户端(懒初始化,连接池复用)"""
129+
global _jwks_client
130+
if _jwks_client is None:
131+
_jwks_client = httpx.AsyncClient(timeout=10)
132+
return _jwks_client
122133

123134

124135
def _resolve_jwks_url() -> str:
@@ -136,26 +147,29 @@ async def _fetch_jwks(jwks_url: str) -> dict:
136147
"""
137148
获取 JWKS(带 TTL 缓存,默认 1 小时刷新一次)。
138149
139-
网络失败时若已有缓存,则返回过期缓存以保证服务可用性。
150+
GW-M3: 网络失败时 fail-closed(拒绝验证)而非返回过期缓存——
151+
密钥轮换后过期缓存会让旧 token 在验证窗口内继续有效(越权窗口)。
152+
asyncio.Lock 防止冷启动多请求并发刷新(缓存击穿)。
140153
"""
141154
global _jwks_cache, _jwks_cache_time
142155
now = time.time()
143156
if _jwks_cache is not None and (now - _jwks_cache_time) < _JWKS_CACHE_TTL:
144157
return _jwks_cache
145-
try:
146-
resp = await asyncio.to_thread(httpx.get, jwks_url, timeout=10)
147-
resp.raise_for_status()
148-
_jwks_cache = resp.json()
149-
_jwks_cache_time = now
150-
logger.info("JWKS 获取成功,缓存已更新: %s", jwks_url)
151-
return _jwks_cache
152-
except Exception as e:
153-
logger.error("JWKS 获取失败: %s, error=%s", jwks_url, str(e))
154-
# 降级:返回过期缓存(若有),避免单次网络抖动导致验签全失败
155-
if _jwks_cache is not None:
156-
logger.warning("JWKS 获取失败,使用过期缓存(可能已过期)")
158+
async with _jwks_lock:
159+
# 双重检查:等待锁期间其他协程可能已刷新缓存
160+
now = time.time()
161+
if _jwks_cache is not None and (now - _jwks_cache_time) < _JWKS_CACHE_TTL:
157162
return _jwks_cache
158-
raise
163+
try:
164+
resp = await _get_jwks_client().get(jwks_url)
165+
resp.raise_for_status()
166+
_jwks_cache = resp.json()
167+
_jwks_cache_time = time.time()
168+
logger.info("JWKS 获取成功,缓存已更新: %s", jwks_url)
169+
return _jwks_cache
170+
except Exception as e:
171+
logger.error("JWKS 获取失败: %s, error=%s", jwks_url, str(e))
172+
raise
159173

160174

161175
def _b64url_to_int(val: str) -> int:

0 commit comments

Comments
 (0)