feat: streaming passthrough for stream=True + first-byte timeout (30s) — distinguish slow upstream (data flowing) vs hung key (no first byte); fix 180s cutoff killing legit 3-8min requests

This commit is contained in:
hmo
2026-08-15 20:22:22 +08:00
parent a7a42448d4
commit 5a736ddaa8
+49 -6
View File
@@ -20,7 +20,7 @@ OpenAI 兼容代理,背后池化 6 个 OpenCode Go API key。
└─ dashboard (:5803) ◄── /api/status
"""
import os, sys, json, time, logging, threading
import os, sys, json, time, logging, threading, socket
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from urllib.request import Request, urlopen, HTTPError
from urllib.error import URLError
@@ -51,6 +51,9 @@ MAX_RETRIES = 6
# 单次上游请求保底超时(秒):上游处理大请求(1MB+)可能要 60-90s+
# 故设宽超时兜底(防 TCP 挂死无限等),正常靠错误分类快速换 key,不依赖此超时
UPSTREAM_REQUEST_TIMEOUT = 180
# 首字节超时(秒):上游连接建立后,30s 内必须返回第一个字节,否则判定该 key 挂起,换下一个 key。
# 这是"区分上游慢 vs 挂"的关键——有数据流=在正常生成,不该中断;无响应=key 有问题,快速切换。
FIRST_BYTE_TIMEOUT = 30
# ── 日志 ────────────────────────────────────────────────────
LOG_DIR = _GATEWAY_DIR / "logs"
@@ -633,7 +636,12 @@ class RouterHandler(BaseHTTPRequestHandler):
self._error_response(502, f"All OCG keys exhausted: {last_error}")
def _forward_request(self, api_key, body, is_stream):
"""转发单次请求到指定 key。返回 {ok, status, error}。"""
"""转发单次请求到指定 key。返回 {ok, status, error}。
stream=True:流式透传——边收上游边发客户端,客户端尽早拿到首 token。
stream=False:完整读取 + 空输出检测(需要完整 JSON)。
两者都做首字节超时(FIRST_BYTE_TIMEOUT):连接建立后 30s 内无响应 = key 挂起,换 key。
"""
try:
req = Request(
f"{OPENCODE_BASE}/chat/completions",
@@ -651,10 +659,43 @@ class RouterHandler(BaseHTTPRequestHandler):
err_body = resp.read().decode("utf-8", errors="replace")[:500]
return {"ok": False, "status": status, "error": f"HTTP {status}: {err_body}"}
# 透传响应 + 空输出检测(2026-08-14 老莫:空输出自动换 key)
# 首字节超时:底层 socket 设 FIRST_BYTE_TIMEOUT
# 30s 内读不到第一个字节 → 判定 key 挂起(socket.timeout 会被捕获并换 key
try:
sock = resp.fp.raw._sock
sock.settimeout(FIRST_BYTE_TIMEOUT)
except Exception:
pass # 拿不到 socket 就算了,靠总超时兜底
if is_stream:
# 流式透传:边收边发,客户端尽早拿到首 token
# 一旦开始有数据流(上游在生成),后续读取不再受首字节超时限制
try:
self.send_response(status)
for k, v in resp.headers.items():
if k.lower() in ("content-type", "access-control-allow-origin"):
self.send_header(k, v)
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
while True:
chunk = resp.read(8192)
if not chunk:
break
try:
sock.settimeout(None) # 已有数据流,解除首字节超时
except Exception:
pass
self.wfile.write(chunk)
self.wfile.flush()
self.wfile.flush()
return {"ok": True}
except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
# 客户端中途断开:数据已发一部分,不算 key 问题
log.info("client disconnected during streaming via %s", api_key[:8])
return {"ok": True} # 响应已开始,视为成功(客户端自行处理断流)
else:
# 非流式:完整读取 + 空输出检测
resp_body = resp.read()
if not is_stream:
# 非流式:解析 JSON 检测空输出
try:
import json as _json
d = _json.loads(resp_body.decode("utf-8", errors="replace"))
@@ -668,7 +709,6 @@ class RouterHandler(BaseHTTPRequestHandler):
has_reasoning = bool(reasoning and str(reasoning).strip())
if (not content or not str(content).strip()) and not has_reasoning:
if finish_reason == "length":
# max_tokens 截断,content 可能为空但属于正常响应
log.warning("content empty but finish_reason=length (truncated), passing through")
else:
log.warning("empty content detected via key, retrying next key")
@@ -684,6 +724,9 @@ class RouterHandler(BaseHTTPRequestHandler):
except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError) as e:
# 客户端断开:不是 key 问题,标记特殊错误码让调用方不重试不冷却
return {"ok": False, "status": -1, "error": f"client_disconnected: {e}"}
except (TimeoutError, socket.timeout) as e:
# 首字节超时或总超时:key 挂起,换下一个 key 重试(不冷却)
return {"ok": False, "status": 0, "error": f"upstream_timeout: {e}"}
except URLError as e:
return {"ok": False, "status": 0, "error": f"URLError: {e.reason}"}
except Exception as e: