Skip to content

Commit 0991ff7

Browse files
committed
修复网关 10605 并发拒绝被误判为登录过期导致的频繁报错
目的: 网关的排队/并发准入拒绝(业务码 10605,携带 isQueued/queueCount/ retryAfterSeconds 等队列元数据)复用 HTTP 401/403 信封与流内 statusCodeValue 帧下发,此前被一律判定为登录过期,触发无意义的 jobToken 强制刷新 + 立即重发(再次撞上同一并发窗口),客户端并发调用时 几乎必现 "HTTP 401 403 {\"code\":\"10605\",...}" 报错。 主要变更: - auth: 新增 BusyError 类型与 10605 解析(HTTP 信封 + 流内帧两条路径), code 兼容字符串/数值两种形态;10605 不再归类为 AuthError - auth: 错误体读取上限 300B → 4KiB,避免长 body 截断致 JSON 解析失败 而回落误判;上游回显消息截断至 512 runes - auth: 流内 busy 检测增加 statusCodeValue 廉价预判,避免对每条 SSE 内容帧做完整 JSON 反序列化 - bridge: openStreamAsync 对 BusyError 改为按 retryAfterSeconds 退避后 重试一次(不再刷新令牌);回退等待提取为可注入变量便于测试 - bridge: HandleChat 新增每 PAT 上游并发闸门(信号量,默认 1,超出本地 排队),从源头避免触发网关限流;可用 QODER_MAX_CONCURRENCY(1-32) 调整 - bridge: 非流式路径忙错误映射为 HTTP 429 + Retry-After(原为 500), 便于 OpenAI 兼容客户端自动退避重试 影响范围: - /v1/chat/completions 同步与流式路径的错误分类、重试行为与状态码映射; /v1/models 及管理面板不受影响 - 新环境变量 QODER_MAX_CONCURRENCY;README 已补充说明 - 行为变化:并发请求默认在本地排队串行执行(原为直冲网关被拒) 测试:auth 新增 5 个用例(数值 code、长 body、消息截断、预判不误伤等), bridge 新增 3 个用例(busy 退避重试且不刷新令牌、429/SSE 错误块映射、 并发闸门实测上游并发=1)。go build/vet/test 全部通过。
1 parent ce75742 commit 0991ff7

5 files changed

Lines changed: 513 additions & 19 deletions

File tree

‎README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ docker run -d -p 10081:10081 -v qoder2api-data:/app/data -e QODER_DATA_PATH=/app
7676
| `QODER_SIGNATURE_SECRET` | 内置值 | 请求签名密钥 |
7777
| `QODER_CHAT_TIMEOUT_SECONDS` | `120` | Chat 响应超时:等待上游开始响应(返回响应头)的最长秒数,范围 1-3600;设置后管理面板中不可修改 |
7878
| `QODER_IDLE_TIMEOUT_SECONDS` | `300` | 流空闲超时:流式响应中持续无数据的最长等待秒数,范围 1-3600(数据持续到达时流不会被中断);设置后管理面板中不可修改 |
79+
| `QODER_MAX_CONCURRENCY` | `1` | 单 PAT 同时发往上游的聊天请求数上限;超出部分本地排队等待。调大会提升吞吐,但超过 Qoder 账号的并发窗口时网关会拒绝请求(业务码 10605) |
7980

8081
> 两个超时也可在管理面板「设置 → 超时配置」中修改,保存后即时生效(无需重启)。响应超时只约束上游「开始响应」,不会截断已经建立的流。
8182

‎auth/auth.go‎

Lines changed: 180 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -337,6 +337,104 @@ func (e *AuthError) Error() string {
337337
return s
338338
}
339339

340+
// --- Gateway busy (queue/concurrency) errors ---
341+
342+
// errBodyCap bounds how much of an upstream error response is read for
343+
// classification and reporting. The former 300-byte cap could truncate a
344+
// 10605 envelope mid-JSON, making it unparseable and misclassifying the
345+
// rejection as an auth failure.
346+
const errBodyCap = 4 << 10 // 4 KiB
347+
348+
// QoderBusyCode is the gateway's business code for queue/admission-control
349+
// rejections. It rides HTTP 401/403 envelopes (and in-stream frames with a
350+
// matching statusCodeValue) even though the session is perfectly valid:
351+
//
352+
// {"code":"10605","message":"{\"isQueued\":false,...,\"retryAfterSeconds\":2,...}"}
353+
//
354+
// Treating it as an auth failure causes a pointless token refresh followed
355+
// by an immediate retry that slams into the same concurrency window.
356+
const QoderBusyCode = "10605"
357+
358+
// BusyError is a gateway queue/concurrency rejection (business code 10605).
359+
// The session is valid; the request was refused by admission control and may
360+
// succeed after RetryAfter.
361+
type BusyError struct {
362+
StatusCode int
363+
Code string
364+
Message string
365+
RetryAfter time.Duration
366+
}
367+
368+
func (e *BusyError) Error() string {
369+
s := fmt.Sprintf("gateway busy (%s): %s", e.Code, e.Message)
370+
if e.RetryAfter > 0 {
371+
s += fmt.Sprintf(" (retry after %s)", e.RetryAfter)
372+
}
373+
return s
374+
}
375+
376+
// busyMeta extracts retryAfterSeconds from the nested JSON message carried
377+
// inside a 10605 payload, if present.
378+
func busyMeta(msg string) time.Duration {
379+
var meta struct {
380+
RetryAfterSeconds int `json:"retryAfterSeconds"`
381+
}
382+
if strings.HasPrefix(strings.TrimSpace(msg), "{") && json.Unmarshal([]byte(msg), &meta) == nil && meta.RetryAfterSeconds > 0 {
383+
return time.Duration(meta.RetryAfterSeconds) * time.Second
384+
}
385+
return 0
386+
}
387+
388+
// maxBusyMessageRunes caps the upstream-provided busy message echoed into
389+
// logs and client-facing error bodies.
390+
const maxBusyMessageRunes = 512
391+
392+
func truncateBusyMessage(s string) string {
393+
r := []rune(s)
394+
if len(r) <= maxBusyMessageRunes {
395+
return s
396+
}
397+
return string(r[:maxBusyMessageRunes]) + "…"
398+
}
399+
400+
// busyCodeOf normalizes a JSON "code" field that may arrive as a string
401+
// ("10605") or a number (10605).
402+
func busyCodeOf(raw json.RawMessage) string {
403+
var s string
404+
if json.Unmarshal(raw, &s) == nil {
405+
return s
406+
}
407+
var n float64
408+
if json.Unmarshal(raw, &n) == nil && n == float64(int64(n)) {
409+
return strconv.FormatInt(int64(n), 10)
410+
}
411+
return ""
412+
}
413+
414+
// busyFromAuthEnvelope parses a 401/403 response body and returns a
415+
// BusyError when it is a queue/concurrency rejection rather than an auth
416+
// failure. Bodies seen in the wild are prefixed with an upstream status
417+
// ("403 {..."), so decoding starts at the first '{'.
418+
func busyFromAuthEnvelope(statusCode int, body []byte) *BusyError {
419+
s := strings.TrimSpace(string(body))
420+
if i := strings.IndexByte(s, '{'); i > 0 {
421+
s = s[i:]
422+
}
423+
var outer struct {
424+
Code json.RawMessage `json:"code"`
425+
Message string `json:"message"`
426+
}
427+
if json.Unmarshal([]byte(s), &outer) != nil || busyCodeOf(outer.Code) != QoderBusyCode {
428+
return nil
429+
}
430+
busy := &BusyError{StatusCode: statusCode, Code: QoderBusyCode}
431+
if outer.Message != "" {
432+
busy.Message = truncateBusyMessage(outer.Message)
433+
busy.RetryAfter = busyMeta(outer.Message)
434+
}
435+
return busy
436+
}
437+
340438
// --- Signature API Client ---
341439

342440
func commonHeaders(machineID, machineToken, machineType, date, sig string) http.Header {
@@ -381,9 +479,12 @@ func postEncoded(ctx context.Context, urlStr string, obj interface{}, machineID,
381479
defer resp.Body.Close()
382480

383481
if resp.StatusCode != 200 {
384-
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 300))
482+
detail, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyCap))
385483
drainBody(resp.Body)
386484
if resp.StatusCode == 401 || resp.StatusCode == 403 {
485+
if busy := busyFromAuthEnvelope(resp.StatusCode, detail); busy != nil {
486+
return nil, busy
487+
}
387488
return nil, &AuthError{StatusCode: resp.StatusCode, Detail: string(detail)}
388489
}
389490
return nil, fmt.Errorf("HTTP %d at %s body=%s", resp.StatusCode, urlStr, string(detail))
@@ -537,9 +638,12 @@ func call(ctx context.Context, sess *SessionContext, method, fullURL string, jso
537638
defer resp.Body.Close()
538639

539640
if resp.StatusCode != 200 {
540-
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 300))
641+
detail, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyCap))
541642
drainBody(resp.Body)
542643
if resp.StatusCode == 401 || resp.StatusCode == 403 {
644+
if busy := busyFromAuthEnvelope(resp.StatusCode, detail); busy != nil {
645+
return nil, busy
646+
}
543647
return nil, &AuthError{StatusCode: resp.StatusCode, Detail: string(detail)}
544648
}
545649
return nil, fmt.Errorf("HTTP %d body=%s", resp.StatusCode, string(detail))
@@ -631,7 +735,9 @@ func isDoneLine(line string) bool {
631735
if payload == "[DONE]" {
632736
return true
633737
}
634-
var env struct{ Body string `json:"body"` }
738+
var env struct {
739+
Body string `json:"body"`
740+
}
635741
if json.Unmarshal([]byte(payload), &env) == nil && env.Body == "[DONE]" {
636742
return true
637743
}
@@ -674,12 +780,12 @@ func (e *inStreamError) Error() string {
674780
// from an SSE data line. Two shapes are recognized:
675781
//
676782
// 1. Envelope frame with an HTTP error status, e.g. (qmodel_preview
677-
// provider outage):
678-
// {"body":"{\"code\":\"provider_error\",\"message\":\"All backends failed\"...}",
679-
// "statusCodeValue":429,"statusCode":"TOO_MANY_REQUESTS"}
783+
// provider outage):
784+
// {"body":"{\"code\":\"provider_error\",\"message\":\"All backends failed\"...}",
785+
// "statusCodeValue":429,"statusCode":"TOO_MANY_REQUESTS"}
680786
//
681787
// 2. Bare success:false gateway error frame, e.g.
682-
// {"success":false,"msgCode":500,"message":"Internal Server Error"}
788+
// {"success":false,"msgCode":500,"message":"Internal Server Error"}
683789
//
684790
// A 200 OK envelope with normal chat/usage/[DONE] bodies returns nil.
685791
func detectInStreamGatewayError(line string) *inStreamError {
@@ -688,9 +794,9 @@ func detectInStreamGatewayError(line string) *inStreamError {
688794
}
689795
payload := strings.TrimSpace(line[len("data:"):])
690796
var env struct {
691-
Body string `json:"body"`
692-
StatusCode string `json:"statusCode"`
693-
StatusValue int `json:"statusCodeValue"`
797+
Body string `json:"body"`
798+
StatusCode string `json:"statusCode"`
799+
StatusValue int `json:"statusCodeValue"`
694800
}
695801
if err := json.Unmarshal([]byte(payload), &env); err != nil {
696802
return nil
@@ -707,10 +813,10 @@ func detectInStreamGatewayError(line string) *inStreamError {
707813
return nil
708814
}
709815
var bare struct {
710-
Success bool `json:"success"`
711-
MsgCode int `json:"msgCode"`
712-
MsgInfo string `json:"msgInfo"`
713-
Message string `json:"message"`
816+
Success bool `json:"success"`
817+
MsgCode int `json:"msgCode"`
818+
MsgInfo string `json:"msgInfo"`
819+
Message string `json:"message"`
714820
}
715821
if err := json.Unmarshal([]byte(payload), &bare); err != nil {
716822
return nil
@@ -775,9 +881,12 @@ func OpenStreamLines(ctx context.Context, sess *SessionContext, fullURL string,
775881
defer resp.Body.Close()
776882

777883
if resp.StatusCode != 200 {
778-
errBody, _ := io.ReadAll(io.LimitReader(resp.Body, 300))
884+
errBody, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyCap))
779885
drainBody(resp.Body)
780886
if resp.StatusCode == 401 || resp.StatusCode == 403 {
887+
if busy := busyFromAuthEnvelope(resp.StatusCode, errBody); busy != nil {
888+
return busy
889+
}
781890
return &AuthError{StatusCode: resp.StatusCode, Detail: string(errBody)}
782891
}
783892
return fmt.Errorf("HTTP %d %s", resp.StatusCode, string(errBody))
@@ -861,6 +970,9 @@ func OpenStreamLines(ctx context.Context, sess *SessionContext, fullURL string,
861970
if e := detectInStreamGatewayError(line); e != nil && gwErr == nil {
862971
gwErr = e
863972
}
973+
if busy := detectInStreamBusyError(line); busy != nil {
974+
return busy
975+
}
864976
isAuthErr, detail := detectInStreamAuthError(line)
865977
if isAuthErr {
866978
return &AuthError{StatusCode: 401, Detail: detail}
@@ -889,7 +1001,57 @@ func OpenStreamLines(ctx context.Context, sess *SessionContext, fullURL string,
8891001
return nil
8901002
}
8911003

1004+
// detectInStreamBusyError reports whether an SSE frame carries the
1005+
// gateway's queue/concurrency rejection (code 10605) under a 401/403
1006+
// statusCodeValue. Such frames look like auth failures but are admission-
1007+
// control refusals; the caller should back off instead of refreshing tokens.
1008+
func detectInStreamBusyError(line string) *BusyError {
1009+
s := strings.TrimSpace(line)
1010+
if !strings.HasPrefix(s, "data:") {
1011+
return nil
1012+
}
1013+
// Cheap pre-filter: busy frames always ride a statusCodeValue envelope.
1014+
// Without this every content delta (potentially large base64 payloads)
1015+
// would pay a full JSON unmarshal per line.
1016+
if !strings.Contains(s, "statusCodeValue") {
1017+
return nil
1018+
}
1019+
var obj map[string]interface{}
1020+
if err := json.Unmarshal([]byte(strings.TrimSpace(s[5:])), &obj); err != nil {
1021+
return nil
1022+
}
1023+
scv, ok := obj["statusCodeValue"].(float64)
1024+
if !ok || (scv != 401 && scv != 403) {
1025+
return nil
1026+
}
1027+
bodyMap, ok := parseBody(obj).(map[string]interface{})
1028+
if !ok {
1029+
return nil
1030+
}
1031+
var code string
1032+
switch c := bodyMap["code"].(type) {
1033+
case string:
1034+
code = c
1035+
case float64:
1036+
if c == float64(int64(c)) {
1037+
code = strconv.FormatInt(int64(c), 10)
1038+
}
1039+
}
1040+
if code != QoderBusyCode {
1041+
return nil
1042+
}
1043+
busy := &BusyError{StatusCode: int(scv), Code: code}
1044+
if msg, _ := bodyMap["message"].(string); msg != "" {
1045+
busy.Message = truncateBusyMessage(msg)
1046+
busy.RetryAfter = busyMeta(msg)
1047+
}
1048+
return busy
1049+
}
1050+
8921051
// detectInStreamAuthError detects auth failures inside the SSE stream.
1052+
// Queue/concurrency rejections (code 10605) share the 401/403 envelope but
1053+
// are NOT auth failures; they are left to detectInStreamBusyError so callers
1054+
// can distinguish "token expired" (refresh + retry) from "back off".
8931055
func detectInStreamAuthError(line string) (bool, string) {
8941056
s := strings.TrimSpace(line)
8951057
if !strings.HasPrefix(s, "data:") {
@@ -899,6 +1061,9 @@ func detectInStreamAuthError(line string) (bool, string) {
8991061
if err := json.Unmarshal([]byte(strings.TrimSpace(s[5:])), &obj); err != nil {
9001062
return false, ""
9011063
}
1064+
if detectInStreamBusyError(line) != nil {
1065+
return false, ""
1066+
}
9021067
scv, ok := obj["statusCodeValue"].(float64)
9031068
if ok && (scv == 401 || scv == 403) {
9041069
return true, fmt.Sprintf("%v %s", scv, extractBodyMessage(obj))

‎auth/auth_test.go‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"io"
88
"net/http"
99
"net/http/httptest"
10+
"strconv"
1011
"strings"
1112
"testing"
1213
"time"
@@ -264,3 +265,87 @@ func TestOpenStreamLinesIdleTimeout(t *testing.T) {
264265
t.Errorf("idle timeout fired too late: %s", elapsed)
265266
}
266267
}
268+
269+
// TestBusyErrorParsing locks the classification of gateway admission-control
270+
// rejections (code 10605): they must surface as BusyError, never AuthError,
271+
// regardless of whether they arrive as an HTTP 401/403 body or an in-stream
272+
// control frame.
273+
func TestBusyErrorParsing(t *testing.T) {
274+
// HTTP envelope: body prefixed with the upstream status ("403 {...").
275+
envBody := []byte(`403 {"code":"10605","message":"{\"isQueued\":false,\"modelKey\":\"qmodel_38max\",\"queueCount\":0,\"queueType\":\"slow\",\"retryAfterSeconds\":2,\"serviceAvailable\":true,\"waitTime\":0}"}`)
276+
busy := busyFromAuthEnvelope(401, envBody)
277+
if busy == nil {
278+
t.Fatal("expected BusyError from 10605 envelope, got nil")
279+
}
280+
if busy.RetryAfter != 2*time.Second {
281+
t.Errorf("RetryAfter = %v, want 2s", busy.RetryAfter)
282+
}
283+
284+
// A genuine auth failure must still classify as AuthError.
285+
if busyFromAuthEnvelope(403, []byte(`{"code":"401","message":"unauthorized"}`)) != nil {
286+
t.Error("non-10605 body must not be a BusyError")
287+
}
288+
289+
// In-stream frame with statusCodeValue 403 + code 10605.
290+
frame := `data: {"body":"{\"code\":\"10605\",\"message\":\"{\\\"retryAfterSeconds\\\":2}\"}","statusCode":"FORBIDDEN","statusCodeValue":403}`
291+
inStream := detectInStreamBusyError(frame)
292+
if inStream == nil {
293+
t.Fatal("expected BusyError from in-stream 10605 frame, got nil")
294+
}
295+
if inStream.RetryAfter != 2*time.Second {
296+
t.Errorf("in-stream RetryAfter = %v, want 2s", inStream.RetryAfter)
297+
}
298+
// ...and must NOT be misread as an auth error.
299+
if isAuth, _ := detectInStreamAuthError(frame); isAuth {
300+
t.Error("10605 frame must not be classified as an auth error")
301+
}
302+
}
303+
304+
// Numeric "code" variants (10605 as a JSON number instead of a string) must
305+
// still classify as busy, on both the HTTP-envelope and in-stream paths.
306+
func TestBusyErrorNumericCode(t *testing.T) {
307+
env := []byte(`{"code":10605,"message":"{\"retryAfterSeconds\":3}"}`)
308+
busy := busyFromAuthEnvelope(403, env)
309+
if busy == nil {
310+
t.Fatal("expected BusyError for numeric code, got nil")
311+
}
312+
if busy.RetryAfter != 3*time.Second {
313+
t.Errorf("RetryAfter = %v, want 3s", busy.RetryAfter)
314+
}
315+
316+
frame := `data: {"body":"{\"code\":10605,\"message\":\"{}\"}","statusCodeValue":401}`
317+
if detectInStreamBusyError(frame) == nil {
318+
t.Fatal("expected BusyError for numeric in-stream code, got nil")
319+
}
320+
}
321+
322+
// A truncated 10605 body (larger than the read cap) must not silently fall
323+
// back to AuthError classification.
324+
func TestBusyErrorFromLongBody(t *testing.T) {
325+
longMsg := `{"isQueued":false,"modelKey":"` + strings.Repeat("m", 4096) + `","retryAfterSeconds":2}`
326+
body := []byte(`{"code":"10605","message":` + strconv.Quote(longMsg) + `}`)
327+
if busyFromAuthEnvelope(401, body) == nil {
328+
t.Fatal("expected BusyError for long 10605 body within read cap, got nil")
329+
}
330+
}
331+
332+
// The upstream-provided message echoed into errors must be length-capped.
333+
func TestBusyErrorMessageCapped(t *testing.T) {
334+
msg := strings.Repeat("x", 10000)
335+
busy := busyFromAuthEnvelope(403, []byte(`{"code":"10605","message":"`+msg+`"}`))
336+
if busy == nil {
337+
t.Fatal("expected BusyError")
338+
}
339+
if got := len([]rune(busy.Message)); got > maxBusyMessageRunes+10 {
340+
t.Errorf("message not capped: %d runes", got)
341+
}
342+
}
343+
344+
// Content-delta frames without a statusCodeValue envelope must skip the busy
345+
// pre-filter cheaply (no false positives either way).
346+
func TestDetectInStreamBusyIgnoresPlainDeltas(t *testing.T) {
347+
line := `data: {"body":"{\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}"}`
348+
if detectInStreamBusyError(line) != nil {
349+
t.Error("plain content delta must not be classified as busy")
350+
}
351+
}

0 commit comments

Comments
 (0)