fix(gateway): 补修 #473/#474 合入后仍在的问题

changelog/unreleased/gzip-heartbeat-followup.md

@@ -0,0 +1,11 @@
+### English
+
+- Gzip request bodies that inflate past the decompressed size limit are now rejected with a clear "request body too large" error (413) instead of being silently truncated and then failing as a JSON syntax error.
+- Body-decoding failures on `POST /v1/messages` (unsupported encoding, corrupt gzip, over-limit inflate) now use the Anthropic error envelope this endpoint already returns for other request errors, not the OpenAI shape.
+- Streaming chat heartbeats still send `X-CLI2API-Account`, `X-Qoder-Account`, and `X-CLI2API-Provider` when the SSE headers are committed early: those values are taken from the account already picked while waiting for the upstream response head. A client disconnect during that heartbeat wait is recorded as canceled, not as a normal stream open.
+
+### 中文
+
+- gzip 请求体解压超过上限时现在会直接拒绝，并返回明确的「request body too large」（413），不再静默截断后再被当成 JSON 语法错误。
+- `POST /v1/messages` 的请求体解码失败（不支持的编码、损坏的 gzip、解压超限）改回该接口原有的 Anthropic 错误格式，而不再使用 OpenAI 形状。
+- 流式对话在提前提交 SSE 响应头时仍会带上 `X-CLI2API-Account`、`X-Qoder-Account` 与 `X-CLI2API-Provider`：这些值来自等待上游响应头期间已经选出的账号。客户端在这段心跳等待中断开时记为取消，而不再记成一次正常开流。

internal/executor/chat.go

@@ -1475,6 +1475,7 @@ func (e ChatExecutor) chatStreamProxy(ctx context.Context, req translate.ChatReq
 			}
 			return StreamResult{}, pickErr
 		}
+		noteStreamRoute(ctx, item.ID, loop.resultProvider(item))
 		poolID := item.ID
 		if !e.acquireInFlight(loop.requestID, poolID) {
 			loop.exclude(item)
@@ -1540,6 +1541,7 @@ func (e ChatExecutor) chatStreamProxy(ctx context.Context, req translate.ChatReq
 		}
 		if account := resp.Header.Get("X-Qoder-Account"); account != "" {
 			item.ID = account
+			noteStreamRoute(ctx, item.ID, loop.resultProvider(item))
 		}
 		if resp.StatusCode >= 300 {
 			body, _ := io.ReadAll(io.LimitReader(resp.Body, maxWorkerResponseBytes))

internal/executor/stream_route_hint.go

@@ -0,0 +1,55 @@
+package executor
+
+import (
+	"context"
+	"sync"
+)
+
+type streamRouteHintKey struct{}
+
+// StreamRouteHint holds the account/provider the executor has already picked
+// for an in-flight stream attempt. The gateway reads it from setStreamHeaders
+// so a heartbeat that commits before ChatStreamProxy returns can still send
+// X-CLI2API-Account and X-CLI2API-Provider.
+type StreamRouteHint struct {
+	mu        sync.Mutex
+	accountID string
+	provider  string
+}
+
+func (h *StreamRouteHint) Set(accountID, provider string) {
+	if h == nil {
+		return
+	}
+	h.mu.Lock()
+	h.accountID = accountID
+	h.provider = provider
+	h.mu.Unlock()
+}
+
+func (h *StreamRouteHint) Get() (accountID, provider string) {
+	if h == nil {
+		return "", ""
+	}
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.accountID, h.provider
+}
+
+func WithStreamRouteHint(ctx context.Context, hint *StreamRouteHint) context.Context {
+	if ctx == nil || hint == nil {
+		return ctx
+	}
+	return context.WithValue(ctx, streamRouteHintKey{}, hint)
+}
+
+func noteStreamRoute(ctx context.Context, accountID, provider string) {
+	if ctx == nil {
+		return
+	}
+	hint, _ := ctx.Value(streamRouteHintKey{}).(*StreamRouteHint)
+	if hint == nil {
+		return
+	}
+	hint.Set(accountID, provider)
+}

internal/gateway/anthropic.go

@@ -21,10 +21,14 @@ func (h *Handler) HandleAnthropicMessages(w http.ResponseWriter, r *http.Request
 	body := decodeRequestBody(w, r, maxChatRequestBytes, maxChatRequestBytes)
 	defer body.Close()
 	if body.err != nil {
-		writeRequestBodyError(w, body.err)
+		writeAnthropicRequestBodyError(w, body.err)
 		return
 	}
 	if err := json.NewDecoder(body).Decode(&source); err != nil {
+		if isRequestBodyTooLarge(err) {
+			writeAnthropicRequestBodyError(w, err)
+			return
+		}
 		writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
 		return
 	}

internal/gateway/openai.go

@@ -31,6 +31,10 @@ func (h *Handler) HandleChatCompletions(w http.ResponseWriter, r *http.Request)
 		return
 	}
 	if err := json.NewDecoder(body).Decode(&req); err != nil {
+		if isRequestBodyTooLarge(err) {
+			writeRequestBodyError(w, err)
+			return
+		}
 		writeErr(w, http.StatusBadRequest, "invalid_request", err.Error())
 		return
 	}
@@ -56,17 +60,37 @@ func (h *Handler) HandleChatCompletions(w http.ResponseWriter, r *http.Request)
 	ctx := execution.Context
 	w.Header().Set("X-Request-Id", requestID)
 
-	setStreamHeaders := func() {
-		w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
-		w.Header().Set("Cache-Control", "no-cache")
-		w.Header().Set("Connection", "keep-alive")
-		w.Header().Set("X-Accel-Buffering", "no")
-	}
-
 	if req.Stream {
-		// Start heartbeating before blocking on the upstream head. A long prompt
-		// can take longer to upload than a proxy's response budget; silence in
-		// that window is what turns a slow turn into an aborted connection.
+		applyStreamIdentityHeaders := func(accountID, provider string) {
+			if accountID != "" {
+				w.Header().Set("X-Qoder-Account", accountID)
+				w.Header().Set("X-CLI2API-Account", accountID)
+			}
+			if provider != "" {
+				w.Header().Set("X-CLI2API-Provider", provider)
+			}
+		}
+
+		// The heartbeat may commit SSE headers while ChatStreamProxy is still
+		// waiting for the upstream head. Account/provider are known as soon as
+		// the executor picks a route, so the timer reads them from this hint
+		// and includes them in the first flush.
+		routeHint := &executor.StreamRouteHint{}
+		ctx = executor.WithStreamRouteHint(ctx, routeHint)
+
+		setStreamHeaders := func() {
+			w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
+			w.Header().Set("Cache-Control", "no-cache")
+			w.Header().Set("Connection", "keep-alive")
+			w.Header().Set("X-Accel-Buffering", "no")
+			accountID, provider := routeHint.Get()
+			applyStreamIdentityHeaders(accountID, firstNonEmpty(provider, providerFilter))
+		}
+
+		// Start heartbeating before blocking on the upstream head. Silence
+		// while waiting for that head (or a long prefill) is what turns a
+		// healthy turn into an aborted connection; the request body has
+		// already been read by this point.
 		heartbeat := newStreamHeartbeat(w)
 		heartbeat.Arm(setStreamHeaders)
 
@@ -76,8 +100,15 @@ func (h *Handler) HandleChatCompletions(w http.ResponseWriter, r *http.Request)
 				// The headers are already on the wire, so the only channel left
 				// is an in-stream error event. Report it the same way a
 				// mid-stream failure is reported.
-				h.finishRequestLog(requestID, started, req, publicModel, upstream.AccountID, firstNonEmpty(upstream.Provider, providerFilter), upstream.Routing, accounts.RequestStatusError, upstream.TTFBMs, nil, err, upstream.AttemptCount, upstream.ReasoningLevel)
-				writeStructuredStreamError(w, err)
+				status := accounts.RequestStatusError
+				if heartbeat.clientGone() != nil || r.Context().Err() != nil {
+					status = accounts.RequestStatusCanceled
+					err = context.Canceled
+				}
+				h.finishRequestLog(requestID, started, req, publicModel, upstream.AccountID, firstNonEmpty(upstream.Provider, providerFilter), upstream.Routing, status, upstream.TTFBMs, nil, err, upstream.AttemptCount, upstream.ReasoningLevel)
+				if status != accounts.RequestStatusCanceled {
+					writeStructuredStreamError(w, err)
+				}
 				return
 			}
 			h.finishRequestLog(requestID, started, req, publicModel, upstream.AccountID, firstNonEmpty(upstream.Provider, providerFilter), upstream.Routing, accounts.RequestStatusError, upstream.TTFBMs, nil, err, upstream.AttemptCount, upstream.ReasoningLevel)
@@ -86,23 +117,19 @@ func (h *Handler) HandleChatCompletions(w http.ResponseWriter, r *http.Request)
 		}
 		committed := heartbeat.Adopt()
 		defer upstream.Response.Body.Close()
+		if heartbeat.clientGone() != nil || r.Context().Err() != nil {
+			// The client left during the heartbeat wait. Do not record this
+			// as a normal stream open; the headers may already be on the
+			// wire, but no upstream tokens were relayed. CommitSession stays
+			// uncalled: it binds only a stream that finished without a client
+			// error, and this one ended because the client left.
+			h.finishRequestLog(requestID, started, req, publicModel, upstream.AccountID, firstNonEmpty(upstream.Provider, providerFilter), upstream.Routing, accounts.RequestStatusCanceled, upstream.TTFBMs, nil, context.Canceled, upstream.AttemptCount, upstream.ReasoningLevel)
+			return
+		}
 		h.finishRequestLog(requestID, started, req, publicModel, upstream.AccountID, firstNonEmpty(upstream.Provider, providerFilter), upstream.Routing, accounts.RequestStatusStreaming, upstream.TTFBMs, nil, nil, upstream.AttemptCount, upstream.ReasoningLevel)
-		if committed {
-			// Headers already sent by the heartbeat; only decorate what a
-			// committed response can still carry.
-			if upstream.AccountID != "" {
-				w.Header().Set("X-Qoder-Account", upstream.AccountID)
-				w.Header().Set("X-CLI2API-Account", upstream.AccountID)
-			}
-		} else {
+		if !committed {
 			setStreamHeaders()
-			if upstream.AccountID != "" {
-				w.Header().Set("X-Qoder-Account", upstream.AccountID)
-				w.Header().Set("X-CLI2API-Account", upstream.AccountID)
-			}
-			if provider := firstNonEmpty(upstream.Provider, providerFilter); provider != "" {
-				w.Header().Set("X-CLI2API-Provider", provider)
-			}
+			applyStreamIdentityHeaders(upstream.AccountID, firstNonEmpty(upstream.Provider, providerFilter))
 			w.WriteHeader(http.StatusOK)
 			if flusher, ok := w.(http.Flusher); ok {
 				flusher.Flush()

internal/gateway/request_body.go

@@ -39,6 +39,11 @@ import (
 // decode. Callers turn it into a 415.
 var ErrUnsupportedContentEncoding = errors.New("unsupported content encoding")
 
+// ErrInflatedBodyTooLarge is returned when a decompressed body exceeds the
+// inflated limit. Unlike io.LimitReader, this is a hard reject: the caller
+// must not continue with a truncated body.
+var ErrInflatedBodyTooLarge = errors.New("decompressed request body too large")
+
 // decodeRequestBody returns a reader for the request body with any supported
 // Content-Encoding removed.
 //
@@ -61,7 +66,7 @@ func decodeRequestBody(w http.ResponseWriter, r *http.Request, wireLimit, inflat
 			return requestBodyReader{err: fmt.Errorf("invalid gzip body: %w", err), raw: body}
 		}
 		return requestBodyReader{
-			Reader: io.LimitReader(gz, inflatedLimit),
+			Reader: &inflatedLimitReader{r: gz, n: inflatedLimit},
 			gz:     gz,
 			raw:    body,
 		}
@@ -70,6 +75,48 @@ func decodeRequestBody(w http.ResponseWriter, r *http.Request, wireLimit, inflat
 	}
 }
 
+// inflatedLimitReader bounds decompressed reads the way http.MaxBytesReader
+// bounds the wire: reading past the limit returns a distinct error instead of
+// a silent EOF, so the JSON decoder cannot treat a truncated object as a
+// syntax typo or, worse, succeed on a prefix that happens to be valid JSON.
+type inflatedLimitReader struct {
+	r io.Reader
+	n int64
+}
+
+func (r *inflatedLimitReader) Read(p []byte) (int, error) {
+	if r.n < 0 {
+		return 0, ErrInflatedBodyTooLarge
+	}
+	if r.n == 0 {
+		return r.rejectIfMore()
+	}
+	if int64(len(p)) > r.n {
+		p = p[:r.n]
+	}
+	n, err := r.r.Read(p)
+	r.n -= int64(n)
+	if r.n == 0 && err == nil {
+		if _, peekErr := r.rejectIfMore(); peekErr != nil {
+			return n, peekErr
+		}
+	}
+	return n, err
+}
+
+func (r *inflatedLimitReader) rejectIfMore() (int, error) {
+	var buf [1]byte
+	n, err := r.r.Read(buf[:])
+	if n > 0 {
+		r.n = -1
+		return 0, ErrInflatedBodyTooLarge
+	}
+	if err == nil {
+		return 0, nil
+	}
+	return 0, err
+}
+
 // requestBodyReader is the decoded body plus the handles needed to release it
 // without leaking the gzip reader.
 type requestBodyReader struct {
@@ -109,14 +156,39 @@ func requestBodyCompressed(r *http.Request) bool {
 	}
 }
 
-// writeRequestBodyError turns a body-decoding failure into the right status.
-// An encoding the gateway cannot decode is a client contract problem (415), not
-// a malformed request; everything else is the existing 400 path so callers keep
-// the message they had before.
+func isRequestBodyTooLarge(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, ErrInflatedBodyTooLarge) {
+		return true
+	}
+	var maxBytes *http.MaxBytesError
+	return errors.As(err, &maxBytes)
+}
+
+// writeRequestBodyError turns a body-decoding failure into the OpenAI error
+// shape. An encoding the gateway cannot decode is a client contract problem
+// (415), not a malformed request; an over-limit body is 413; everything else
+// is the existing 400 path so callers keep the message they had before.
 func writeRequestBodyError(w http.ResponseWriter, err error) {
+	status, code, msg := classifyRequestBodyError(err)
+	writeErr(w, status, code, msg)
+}
+
+// writeAnthropicRequestBodyError is the /v1/messages counterpart: same status
+// and message, but the Anthropic error envelope this endpoint already uses.
+func writeAnthropicRequestBodyError(w http.ResponseWriter, err error) {
+	status, _, msg := classifyRequestBodyError(err)
+	writeAnthropicError(w, status, "invalid_request_error", msg)
+}
+
+func classifyRequestBodyError(err error) (status int, code, msg string) {
 	if errors.Is(err, ErrUnsupportedContentEncoding) {
-		writeErr(w, http.StatusUnsupportedMediaType, "unsupported_content_encoding", err.Error())
-		return
+		return http.StatusUnsupportedMediaType, "unsupported_content_encoding", err.Error()
+	}
+	if isRequestBodyTooLarge(err) {
+		return http.StatusRequestEntityTooLarge, "invalid_request", "request body too large"
 	}
-	writeErr(w, http.StatusBadRequest, "invalid_request", err.Error())
+	return http.StatusBadRequest, "invalid_request", err.Error()
 }