zizi-al/sub2api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(gateway): cap upstream error body reads

Browse Source
This commit is contained in:
name 2026-05-30 19:51:03 +08:00
parent 09af6ebd40
commit 34de99ee0e
18 changed files with 392 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
backend/internal/service/antigravity_gateway_service.go
View File

@ -662,7 +662,7 @@ urlFallbackLoop:
// 统一处理错误响应
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
if overagesInjected && shouldMarkCreditsExhausted(resp, respBody, nil) {
@ -875,6 +875,22 @@ type AntigravityGatewayService struct {
internal500Cache Internal500CounterCache // INTERNAL 500 渐进惩罚计数器
}
func (s *AntigravityGatewayService) upstreamErrorBodyReadLimit() int64 {
limit := gatewayUpstreamErrorBodyReadLimit
if s != nil && s.settingService != nil && s.settingService.cfg != nil && s.settingService.cfg.Gateway.LogUpstreamErrorBody && s.settingService.cfg.Gateway.LogUpstreamErrorBodyMaxBytes > int(limit) {
limit = int64(s.settingService.cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
}
return limit
}
func (s *AntigravityGatewayService) readUpstreamErrorBody(resp *http.Response) []byte {
if resp == nil || resp.Body == nil {
return nil
}
body, _ := io.ReadAll(io.LimitReader(resp.Body, s.upstreamErrorBodyReadLimit()))
return body
}
func NewAntigravityGatewayService(
accountRepo AccountRepository,
cache GatewayCache,
@ -1090,7 +1106,7 @@ func (s *AntigravityGatewayService) TestConnection(ctx context.Context, account
}
defer func() { _ = result.resp.Body.Close() }()
respBody, err := io.ReadAll(io.LimitReader(result.resp.Body, 2<<20))
respBody, err := io.ReadAll(io.LimitReader(result.resp.Body, s.upstreamErrorBodyReadLimit()))
if err != nil {
return nil, fmt.Errorf("读取响应失败: %w", err)
}
@ -1427,7 +1443,7 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
// 优先检测 thinking block 的 signature 相关错误（400）并重试一次：
// Antigravity /v1internal 链路在部分场景会对 thought/thinking signature 做严格校验，
@ -1622,7 +1638,7 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
resp = retryResp
respBody = nil
} else {
retryBody, _ := io.ReadAll(io.LimitReader(retryResp.Body, 2<<20))
retryBody := s.readUpstreamErrorBody(retryResp)
_ = retryResp.Body.Close()
respBody = retryBody
resp = &http.Response{
@ -2189,7 +2205,7 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
// 处理错误响应
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
contentType := resp.Header.Get("Content-Type")
// 尽早关闭原始响应体，释放连接；后续逻辑仍可能需要读取 body，因此用内存副本重新包装。
_ = resp.Body.Close()
@ -2270,7 +2286,7 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
if retryResp.StatusCode < 400 {
resp = retryResp
} else {
retryRespBody, _ := io.ReadAll(io.LimitReader(retryResp.Body, 2<<20))
retryRespBody := s.readUpstreamErrorBody(retryResp)
_ = retryResp.Body.Close()
retryOpsBody := retryRespBody
if retryUnwrapped, unwrapErr := s.unwrapV1InternalResponse(retryRespBody); unwrapErr == nil && len(retryUnwrapped) > 0 {
@ -4252,7 +4268,7 @@ func (s *AntigravityGatewayService) ForwardUpstream(ctx context.Context, c *gin.
// 处理错误响应
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
// 429 错误时标记账号限流
if resp.StatusCode == http.StatusTooManyRequests {

9
backend/internal/service/antigravity_gateway_service_test.go
View File

@ -42,6 +42,15 @@ func newAntigravityTestService(cfg *config.Config) *AntigravityGatewayService {
}
}
func TestAntigravityUpstreamErrorBodyReadLimit_RespectsDiagnosticLimit(t *testing.T) {
svc := newAntigravityTestService(&config.Config{Gateway: config.GatewayConfig{
LogUpstreamErrorBody: true,
LogUpstreamErrorBodyMaxBytes: int(gatewayUpstreamErrorBodyReadLimit) + 1024,
}})
require.Equal(t, int64(svc.settingService.cfg.Gateway.LogUpstreamErrorBodyMaxBytes), svc.upstreamErrorBodyReadLimit())
}
func TestStripSignatureSensitiveBlocksFromClaudeRequest(t *testing.T) {
req := &antigravity.ClaudeRequest{
Model: "claude-sonnet-4-5",

2
backend/internal/service/gateway_forward_as_chat_completions.go
View File

@ -148,7 +148,7 @@ func (s *GatewayService) ForwardAsChatCompletions(
// 12. Handle error response with failover
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))

2
backend/internal/service/gateway_forward_as_responses.go
View File

@ -145,7 +145,7 @@ func (s *GatewayService) ForwardAsResponses(
// 12. Handle error response with failover
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))

62
backend/internal/service/gateway_service.go
View File

@ -23,6 +23,7 @@ import (
"sync/atomic"
"syscall"
"time"
"unsafe"
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/Wei-Shaw/sub2api/internal/pkg/claude"
@ -56,6 +57,8 @@ const (
defaultModelsListCacheTTL = 15 * time.Second
postUsageBillingTimeout = 15 * time.Second
debugGatewayBodyEnv = "SUB2API_DEBUG_GATEWAY_BODY"
// 上游错误体只需要提取错误 JSON/日志摘要，默认 512KiB 避免错误风暴叠加大请求体。
gatewayUpstreamErrorBodyReadLimit int64 = 512 << 10
)
const (
@ -833,8 +836,16 @@ func (s *GatewayService) extractCacheableContent(parsed *ParsedRequest) string {
return systemText
}
func parseRawJSONView(raw []byte) gjson.Result {
if len(raw) == 0 {
return gjson.Result{}
}
// 这里只做同步只读解析，避免 gjson.ParseBytes 为大 messages/contents 复制整段 raw。
return gjson.Parse(*(*string)(unsafe.Pointer(&raw)))
}
func extractTextFromSystemRaw(raw []byte) string {
system := gjson.ParseBytes(raw)
system := parseRawJSONView(raw)
switch system.Type {
case gjson.String:
return system.String()
@ -880,7 +891,7 @@ func appendMessageTextsFromRaw(builder *strings.Builder, raw []byte) {
if builder == nil || len(raw) == 0 {
return
}
messages := gjson.ParseBytes(raw)
messages := parseRawJSONView(raw)
if !messages.IsArray() {
return
}
@ -903,7 +914,7 @@ func appendMessageTextsFromRaw(builder *strings.Builder, raw []byte) {
}
func extractCacheableTextFromSystemRaw(raw []byte) string {
system := gjson.ParseBytes(raw)
system := parseRawJSONView(raw)
if !system.IsArray() {
return ""
}
@ -920,7 +931,7 @@ func extractCacheableTextFromSystemRaw(raw []byte) string {
}
func extractCacheableTextFromMessagesRaw(raw []byte) string {
messages := gjson.ParseBytes(raw)
messages := parseRawJSONView(raw)
if !messages.IsArray() {
return ""
}
@ -4676,7 +4687,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
// 优先检测thinking block签名错误（400）并重试一次
if resp.StatusCode == 400 {
respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, readErr := s.readUpstreamErrorBody(resp)
if readErr == nil {
_ = resp.Body.Close()
@ -4739,7 +4750,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
break
}
retryRespBody, retryReadErr := io.ReadAll(io.LimitReader(retryResp.Body, 2<<20))
retryRespBody, retryReadErr := s.readUpstreamErrorBody(retryResp)
_ = retryResp.Body.Close()
if retryReadErr == nil && retryResp.StatusCode == 400 && s.isSignatureErrorPattern(ctx, account, retryRespBody) {
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
@ -4889,7 +4900,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
break
}
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
Platform: account.Platform,
@ -4936,7 +4947,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
// 处理重试耗尽的情况
if resp.StatusCode >= 400 && s.shouldRetryUpstreamError(account, resp.StatusCode) {
if s.shouldFailoverUpstreamError(resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
@ -4971,7 +4982,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
// 处理可切换账号的错误
if resp.StatusCode >= 400 && s.shouldFailoverUpstreamError(resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
@ -5003,7 +5014,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
if resp.StatusCode >= 400 {
// 可选：对部分 400 触发 failover（默认关闭以保持语义）
if resp.StatusCode == 400 && s.cfg != nil && s.cfg.Gateway.FailoverOn400 {
respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, readErr := s.readUpstreamErrorBody(resp)
if readErr != nil {
// ReadAll failed, fall back to normal error handling without consuming the stream
return s.handleErrorResponse(ctx, resp, c, account, reqModel)
@ -5221,7 +5232,7 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
break
}
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
Platform: account.Platform,
@ -5259,7 +5270,7 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
if resp.StatusCode >= 400 && s.shouldRetryUpstreamError(account, resp.StatusCode) {
if s.shouldFailoverUpstreamError(resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
@ -5293,7 +5304,7 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
}
if resp.StatusCode >= 400 && s.shouldFailoverUpstreamError(resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
@ -6011,7 +6022,7 @@ func (s *GatewayService) executeBedrockUpstream(
break
}
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
appendOpsUpstreamError(c, OpsUpstreamErrorEvent{
Platform: account.Platform,
@ -6056,7 +6067,7 @@ func (s *GatewayService) handleBedrockUpstreamErrors(
// retry exhausted + failover
if s.shouldRetryUpstreamError(account, resp.StatusCode) {
if s.shouldFailoverUpstreamError(resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
@ -6083,7 +6094,7 @@ func (s *GatewayService) handleBedrockUpstreamErrors(
// non-retryable failover
if s.shouldFailoverUpstreamError(resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
@ -7226,8 +7237,19 @@ func isCountTokensUnsupported404(statusCode int, body []byte) bool {
return strings.Contains(msg, "count_tokens") && strings.Contains(msg, "not found")
}
func (s *GatewayService) readUpstreamErrorBody(resp *http.Response) ([]byte, error) {
if resp == nil || resp.Body == nil {
return nil, nil
}
limit := gatewayUpstreamErrorBodyReadLimit
if s != nil && s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody && s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes > int(limit) {
limit = int64(s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
}
return io.ReadAll(io.LimitReader(resp.Body, limit))
}
func (s *GatewayService) handleErrorResponse(ctx context.Context, resp *http.Response, c *gin.Context, account *Account, requestedModel ...string) (*ForwardResult, error) {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body, _ := s.readUpstreamErrorBody(resp)
// 调试日志：打印上游错误响应
logger.LegacyPrintf("service.gateway", "[Forward] Upstream error (non-retryable): Account=%d(%s) Status=%d RequestID=%s Body=%s",
@ -7380,7 +7402,7 @@ func (s *GatewayService) handleErrorResponse(ctx context.Context, resp *http.Res
}
func (s *GatewayService) handleRetryExhaustedSideEffects(ctx context.Context, resp *http.Response, account *Account) {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body, _ := s.readUpstreamErrorBody(resp)
statusCode := resp.StatusCode
// OAuth/Setup Token 账号的 403：标记账号异常
@ -7394,7 +7416,7 @@ func (s *GatewayService) handleRetryExhaustedSideEffects(ctx context.Context, re
}
func (s *GatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account, requestedModel ...string) {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body, _ := s.readUpstreamErrorBody(resp)
if len(requestedModel) > 0 {
s.rateLimitService.HandleUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, requestedModel[0])
return
@ -7407,7 +7429,7 @@ func (s *GatewayService) handleFailoverSideEffects(ctx context.Context, resp *ht
// API Key 未配置错误码：仅返回错误，不标记账号
func (s *GatewayService) handleRetryExhaustedError(ctx context.Context, resp *http.Response, c *gin.Context, account *Account) (*ForwardResult, error) {
// Capture upstream error body before side-effects consume the stream.
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody, _ := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))

214
backend/internal/service/gateway_service_benchmark_test.go
View File

@ -4,9 +4,14 @@ import (
"strconv"
"strings"
"testing"
"github.com/Wei-Shaw/sub2api/internal/domain"
)
var benchmarkStringSink string
var (
benchmarkStringSink string
benchmarkIntSink int
)
// BenchmarkGenerateSessionHash_Metadata 关注 JSON 解析与正则匹配开销。
func BenchmarkGenerateSessionHash_Metadata(b *testing.B) {
@ -23,6 +28,121 @@ func BenchmarkGenerateSessionHash_Metadata(b *testing.B) {
}
}
func BenchmarkParseGatewayRequest_LargeAnthropicMessages(b *testing.B) {
for _, size := range benchmarkBodySizes() {
b.Run(size.name, func(b *testing.B) {
body := buildLargeAnthropicMessagesBody(size.bytes, false)
b.SetBytes(int64(len(body)))
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformAnthropic)
if err != nil {
b.Fatalf("解析 Anthropic 请求失败: %v", err)
}
benchmarkIntSink = len(parsed.MessagesRaw())
}
})
}
}
func BenchmarkParseGatewayRequest_LargeGeminiContents(b *testing.B) {
for _, size := range benchmarkBodySizes() {
b.Run(size.name, func(b *testing.B) {
body := buildLargeGeminiContentsBody(size.bytes)
b.SetBytes(int64(len(body)))
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformGemini)
if err != nil {
b.Fatalf("解析 Gemini 请求失败: %v", err)
}
benchmarkIntSink = len(parsed.MessagesRaw())
}
})
}
}
func BenchmarkGenerateSessionHash_LargeAnthropicMessages(b *testing.B) {
svc := &GatewayService{}
for _, size := range benchmarkBodySizes() {
b.Run(size.name, func(b *testing.B) {
body := buildLargeAnthropicMessagesBody(size.bytes, true)
parsed, err := ParseGatewayRequest(NewRequestBodyRef(body), domain.PlatformAnthropic)
if err != nil {
b.Fatalf("解析请求失败: %v", err)
}
b.SetBytes(int64(len(body)))
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
benchmarkStringSink = svc.GenerateSessionHash(parsed)
}
})
}
}
func BenchmarkOpenAIResponses_LargeInputMeta(b *testing.B) {
for _, size := range benchmarkBodySizes() {
b.Run(size.name, func(b *testing.B) {
body := buildLargeOpenAIResponsesBody(size.bytes)
b.SetBytes(int64(len(body)))
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
model, stream, promptCacheKey := extractOpenAIRequestMetaFromBody(body)
benchmarkStringSink = model + promptCacheKey
if stream {
benchmarkIntSink++
}
}
})
}
}
func BenchmarkOpenAIResponses_LargeInputDecodeMap(b *testing.B) {
for _, size := range benchmarkBodySizes() {
b.Run(size.name, func(b *testing.B) {
body := buildLargeOpenAIResponsesBody(size.bytes)
b.SetBytes(int64(len(body)))
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
reqBody, err := getOpenAIRequestBodyMap(nil, body)
if err != nil {
b.Fatalf("解析 OpenAI 请求失败: %v", err)
}
benchmarkIntSink = len(reqBody)
}
})
}
}
func BenchmarkOpenAIResponses_LargeInputFunctionCallValidation(b *testing.B) {
for _, size := range benchmarkBodySizes() {
b.Run(size.name, func(b *testing.B) {
body := buildLargeOpenAIResponsesToolContinuationBody(size.bytes)
b.SetBytes(int64(len(body)))
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
validation := ValidateFunctionCallOutputContextBytes(body)
if !validation.HasFunctionCallOutput || !validation.HasItemReferenceForAllCallIDs {
b.Fatalf("工具续链校验结果异常: %+v", validation)
}
benchmarkIntSink++
}
})
}
}
// BenchmarkExtractCacheableContent_System 关注字符串拼接路径的性能。
func BenchmarkExtractCacheableContent_System(b *testing.B) {
svc := &GatewayService{}
@ -34,6 +154,21 @@ func BenchmarkExtractCacheableContent_System(b *testing.B) {
}
}
func benchmarkBodySizes() []struct {
name string
bytes int
} {
return []struct {
name string
bytes int
}{
{name: "4MB", bytes: 4 << 20},
{name: "8MB", bytes: 8 << 20},
{name: "16MB", bytes: 16 << 20},
{name: "32MB", bytes: 32 << 20},
}
}
func buildSystemCacheableRequest(parts int) *ParsedRequest {
var builder strings.Builder
builder.WriteString(`{"system":[`)
@ -52,3 +187,80 @@ func buildSystemCacheableRequest(parts int) *ParsedRequest {
}
return parsed
}
func buildLargeAnthropicMessagesBody(targetBytes int, includeCacheControl bool) []byte {
var builder strings.Builder
builder.Grow(targetBytes + 1024)
builder.WriteString(`{"model":"claude-sonnet-4-5","stream":true,"system":[{"type":"text","text":"system seed"}],"messages":[`)
for i := 0; builder.Len() < targetBytes; i++ {
if i > 0 {
builder.WriteByte(',')
}
builder.WriteString(`{"role":"user","content":[{"type":"text","text":"`)
builder.WriteString(strings.Repeat("anthropic payload ", 64))
builder.WriteString(strconv.Itoa(i))
builder.WriteByte('"')
if includeCacheControl && i%32 == 0 {
builder.WriteString(`,"cache_control":{"type":"ephemeral"}`)
}
builder.WriteString(`}]}`)
}
builder.WriteString(`]}`)
return []byte(builder.String())
}
func buildLargeGeminiContentsBody(targetBytes int) []byte {
var builder strings.Builder
builder.Grow(targetBytes + 1024)
builder.WriteString(`{"model":"gemini-2.5-pro","systemInstruction":{"parts":[{"text":"system seed"}]},"contents":[`)
for i := 0; builder.Len() < targetBytes; i++ {
if i > 0 {
builder.WriteByte(',')
}
builder.WriteString(`{"role":"user","parts":[{"text":"`)
builder.WriteString(strings.Repeat("gemini payload ", 64))
builder.WriteString(strconv.Itoa(i))
builder.WriteString(`"}]}`)
}
builder.WriteString(`]}`)
return []byte(builder.String())
}
func buildLargeOpenAIResponsesBody(targetBytes int) []byte {
var builder strings.Builder
builder.Grow(targetBytes + 1024)
builder.WriteString(`{"model":"gpt-5.4","stream":true,"prompt_cache_key":"session-benchmark","input":[`)
for i := 0; builder.Len() < targetBytes; i++ {
if i > 0 {
builder.WriteByte(',')
}
builder.WriteString(`{"type":"message","role":"user","content":[{"type":"input_text","text":"`)
builder.WriteString(strings.Repeat("openai responses payload ", 48))
builder.WriteString(strconv.Itoa(i))
builder.WriteString(`"}]}`)
}
builder.WriteString(`],"tools":[{"type":"function","name":"lookup","parameters":{"type":"object","properties":{"query":{"type":"string"}}}}]}`)
return []byte(builder.String())
}
func buildLargeOpenAIResponsesToolContinuationBody(targetBytes int) []byte {
var builder strings.Builder
builder.Grow(targetBytes + 1024)
builder.WriteString(`{"model":"gpt-5.4","stream":true,"previous_response_id":"resp_benchmark","input":[`)
for i := 0; builder.Len() < targetBytes; i++ {
if i > 0 {
builder.WriteByte(',')
}
callID := "call_" + strconv.Itoa(i)
builder.WriteString(`{"type":"item_reference","id":"`)
builder.WriteString(callID)
builder.WriteString(`"},{"type":"function_call_output","call_id":"`)
builder.WriteString(callID)
builder.WriteString(`","output":"`)
builder.WriteString(strings.Repeat("tool output payload ", 48))
builder.WriteString(strconv.Itoa(i))
builder.WriteString(`"}`)
}
builder.WriteString(`]}`)
return []byte(builder.String())
}

4
backend/internal/service/gemini_chat_completions_compat_service.go
View File

@ -151,7 +151,7 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
}
if resp.StatusCode >= 400 && s.shouldRetryGeminiUpstreamError(account, resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
if resp.StatusCode == http.StatusForbidden && isGeminiInsufficientScope(resp.Header, respBody) {
resp = &http.Response{
@ -207,7 +207,7 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
reasoningEffort := extractCCReasoningEffortFromBody(originalChatBody)
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
s.handleGeminiUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
evBody := unwrapIfNeeded(account.Type == AccountTypeOAuth, respBody)

24
backend/internal/service/gemini_messages_compat_service.go
View File

@ -56,6 +56,18 @@ type GeminiMessagesCompatService struct {
responseHeaderFilter *responseheaders.CompiledHeaderFilter
}
func (s *GeminiMessagesCompatService) readUpstreamErrorBody(resp *http.Response) []byte {
if resp == nil || resp.Body == nil {
return nil
}
limit := gatewayUpstreamErrorBodyReadLimit
if s != nil && s.cfg != nil && s.cfg.Gateway.LogUpstreamErrorBody && s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes > int(limit) {
limit = int64(s.cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
}
body, _ := io.ReadAll(io.LimitReader(resp.Body, limit))
return body
}
func NewGeminiMessagesCompatService(
accountRepo AccountRepository,
groupRepo GroupRepository,
@ -789,7 +801,7 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
// Special-case: signature/thought_signature validation errors are not transient, but may be fixed by
// downgrading Claude thinking/tool history to plain text (conservative two-stage retry).
if resp.StatusCode == http.StatusBadRequest && signatureRetryStage < 2 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
if isGeminiSignatureRelatedError(respBody) {
@ -860,7 +872,7 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
}
if resp.StatusCode >= 400 && s.shouldRetryGeminiUpstreamError(account, resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
// Don't treat insufficient-scope as transient.
if resp.StatusCode == 403 && isGeminiInsufficientScope(resp.Header, respBody) {
@ -919,7 +931,7 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
// 统一错误策略：自定义错误码 + 临时不可调度
if s.rateLimitService != nil {
switch s.rateLimitService.CheckErrorPolicy(ctx, account, resp.StatusCode, respBody) {
@ -1329,7 +1341,7 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
}
if resp.StatusCode >= 400 && s.shouldRetryGeminiUpstreamError(account, resp.StatusCode) {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
// Don't treat insufficient-scope as transient.
if resp.StatusCode == 403 && isGeminiInsufficientScope(resp.Header, respBody) {
@ -1410,7 +1422,7 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
isOAuth := account.Type == AccountTypeOAuth
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
// Best-effort fallback for OAuth tokens missing AI Studio scopes when calling countTokens.
// This avoids Gemini SDKs failing hard during preflight token counting.
// Checked before error policy so it always works regardless of custom error codes.
@ -1619,7 +1631,7 @@ func (s *GeminiMessagesCompatService) checkErrorPolicyInLoop(
if resp.StatusCode < 400 || s.rateLimitService == nil {
return false, resp
}
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
rebuilt = &http.Response{
StatusCode: resp.StatusCode,

2
backend/internal/service/openai_embeddings.go
View File

@ -104,7 +104,7 @@ func (s *OpenAIGatewayService) ForwardEmbeddings(
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))

2
backend/internal/service/openai_gateway_chat_completions.go
View File

@ -241,7 +241,7 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions(
// 8. Handle error response with failover
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))

2
backend/internal/service/openai_gateway_chat_completions_raw.go
View File

@ -183,7 +183,7 @@ func (s *OpenAIGatewayService) forwardAsRawChatCompletions(
// 7. Handle error response with failover
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))

2
backend/internal/service/openai_gateway_messages.go
View File

@ -300,7 +300,7 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic(
// 8. Handle error response with failover
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))

2
backend/internal/service/openai_gateway_responses_chat_fallback.go
View File

@ -163,7 +163,7 @@ func (s *OpenAIGatewayService) forwardResponsesViaRawChatCompletions(
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))

34
backend/internal/service/openai_gateway_service.go
View File

@ -49,6 +49,8 @@ const (
// OpenAI WS Mode 失败后的重连次数上限（不含首次尝试）。
// 与 Codex 客户端保持一致：失败后最多重连 5 次。
openAIWSReconnectRetryLimit = 5
// 上游错误体只需要提取错误 JSON/日志摘要，默认 512KiB 避免错误风暴叠加大请求体。
openAIUpstreamErrorBodyReadLimit int64 = 512 << 10
// OpenAI WS Mode 重连退避默认值（可由配置覆盖）。
openAIWSRetryBackoffInitialDefault = 120 * time.Millisecond
openAIWSRetryBackoffMaxDefault = 2 * time.Second
@ -2296,8 +2298,28 @@ func marshalOpenAIUpstreamJSON(v any) ([]byte, error) {
return out, nil
}
func openAIUpstreamErrorBodyReadLimitForConfig(cfg *config.Config) int64 {
limit := openAIUpstreamErrorBodyReadLimit
if cfg != nil && cfg.Gateway.LogUpstreamErrorBody && cfg.Gateway.LogUpstreamErrorBodyMaxBytes > int(limit) {
limit = int64(cfg.Gateway.LogUpstreamErrorBodyMaxBytes)
}
return limit
}
func (s *OpenAIGatewayService) readUpstreamErrorBody(resp *http.Response) []byte {
if resp == nil || resp.Body == nil {
return nil
}
cfg := (*config.Config)(nil)
if s != nil {
cfg = s.cfg
}
body, _ := io.ReadAll(io.LimitReader(resp.Body, openAIUpstreamErrorBodyReadLimitForConfig(cfg)))
return body
}
func (s *OpenAIGatewayService) handleFailoverSideEffects(ctx context.Context, resp *http.Response, account *Account, requestedModel ...string) {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body := s.readUpstreamErrorBody(resp)
if len(requestedModel) > 0 {
s.handleOpenAIAccountUpstreamError(ctx, account, resp.StatusCode, resp.Header, body, requestedModel[0])
return
@ -3045,7 +3067,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
// Handle error response
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
@ -3563,7 +3585,7 @@ func (s *OpenAIGatewayService) handleFailoverErrorResponsePassthrough(
account *Account,
requestBody []byte,
) error {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body := s.readUpstreamErrorBody(resp)
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
@ -3605,7 +3627,7 @@ func (s *OpenAIGatewayService) handleErrorResponsePassthrough(
account *Account,
requestBody []byte,
) error {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body := s.readUpstreamErrorBody(resp)
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
@ -4298,7 +4320,7 @@ func (s *OpenAIGatewayService) handleErrorResponse(
requestBody []byte,
requestedModel ...string,
) (*OpenAIForwardResult, error) {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body := s.readUpstreamErrorBody(resp)
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body))
upstreamMsg = sanitizeUpstreamErrorMessage(upstreamMsg)
@ -4459,7 +4481,7 @@ func (s *OpenAIGatewayService) handleCompatErrorResponse(
writeError compatErrorWriter,
requestedModel ...string,
) (*OpenAIForwardResult, error) {
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
body := s.readUpstreamErrorBody(resp)
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(body))
if upstreamMsg == "" {

23
backend/internal/service/openai_images.go
View File

@ -622,7 +622,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey(
return nil, fmt.Errorf("upstream request failed: %s", safeErr)
}
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))
@ -1276,21 +1276,22 @@ func resolveOpenAIImageBytes(
headers http.Header,
conversationID string,
pointer openAIImagePointerInfo,
errorBodyReadLimit int64,
) ([]byte, error) {
if normalized := normalizeOpenAIImageBase64(pointer.B64JSON); normalized != "" {
return base64.StdEncoding.DecodeString(normalized)
}
if downloadURL := strings.TrimSpace(pointer.DownloadURL); downloadURL != "" {
return downloadOpenAIImageBytes(ctx, client, headers, downloadURL)
return downloadOpenAIImageBytes(ctx, client, headers, downloadURL, errorBodyReadLimit)
}
if strings.TrimSpace(pointer.Pointer) == "" {
return nil, fmt.Errorf("image asset is missing pointer, url, and base64 data")
}
downloadURL, err := fetchOpenAIImageDownloadURL(ctx, client, headers, conversationID, pointer.Pointer)
downloadURL, err := fetchOpenAIImageDownloadURL(ctx, client, headers, conversationID, pointer.Pointer, errorBodyReadLimit)
if err != nil {
return nil, err
}
return downloadOpenAIImageBytes(ctx, client, headers, downloadURL)
return downloadOpenAIImageBytes(ctx, client, headers, downloadURL, errorBodyReadLimit)
}
func normalizeOpenAIImageBase64(raw string) string {
@ -1395,6 +1396,7 @@ func fetchOpenAIImageDownloadURL(
headers http.Header,
conversationID string,
pointer string,
errorBodyReadLimit int64,
) (string, error) {
url := ""
allowConversationRetry := false
@ -1425,7 +1427,7 @@ func fetchOpenAIImageDownloadURL(
} else if resp.IsSuccessState() && strings.TrimSpace(result.DownloadURL) != "" {
return strings.TrimSpace(result.DownloadURL), nil
} else {
statusErr := newOpenAIImageStatusError(resp, "fetch image download url failed")
statusErr := newOpenAIImageStatusError(resp, "fetch image download url failed", errorBodyReadLimit)
if !allowConversationRetry || !isOpenAIImageTransientConversationNotFoundError(statusErr) {
return "", statusErr
}
@ -1450,7 +1452,7 @@ func fetchOpenAIImageDownloadURL(
return "", lastErr
}
func downloadOpenAIImageBytes(ctx context.Context, client *req.Client, headers http.Header, downloadURL string) ([]byte, error) {
func downloadOpenAIImageBytes(ctx context.Context, client *req.Client, headers http.Header, downloadURL string, errorBodyReadLimit int64) ([]byte, error) {
request := client.R().
SetContext(ctx).
DisableAutoReadResponse()
@ -1478,7 +1480,7 @@ func downloadOpenAIImageBytes(ctx context.Context, client *req.Client, headers h
}
}()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return nil, newOpenAIImageStatusError(resp, "download image bytes failed")
return nil, newOpenAIImageStatusError(resp, "download image bytes failed", errorBodyReadLimit)
}
return io.ReadAll(io.LimitReader(resp.Body, openAIImageMaxDownloadBytes))
}
@ -1505,7 +1507,7 @@ func (e *openAIImageStatusError) Error() string {
return "openai image backend request failed"
}
func newOpenAIImageStatusError(resp *req.Response, fallback string) error {
func newOpenAIImageStatusError(resp *req.Response, fallback string, errorBodyReadLimit int64) error {
if resp == nil {
if strings.TrimSpace(fallback) == "" {
fallback = "openai image backend request failed"
@ -1526,7 +1528,10 @@ func newOpenAIImageStatusError(resp *req.Response, fallback string) error {
requestURL = resp.Request.URL.String()
}
if resp.Body != nil {
body, _ = io.ReadAll(io.LimitReader(resp.Body, 2<<20))
if errorBodyReadLimit <= 0 {
errorBodyReadLimit = openAIUpstreamErrorBodyReadLimit
}
body, _ = io.ReadAll(io.LimitReader(resp.Body, errorBodyReadLimit))
_ = resp.Body.Close()
}
}

2
backend/internal/service/openai_images_responses.go
View File

@ -1172,7 +1172,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth(
return nil, fmt.Errorf("upstream request failed: %s", safeErr)
}
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
respBody := s.readUpstreamErrorBody(resp)
_ = resp.Body.Close()
resp.Body = io.NopCloser(bytes.NewReader(respBody))
upstreamMsg := strings.TrimSpace(extractUpstreamErrorMessage(respBody))

31
backend/internal/service/openai_images_test.go
View File

@ -4,6 +4,7 @@ import (
"bytes"
"context"
"errors"
"fmt"
"io"
"mime/multipart"
"net/http"
@ -14,6 +15,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/config"
"github.com/gin-gonic/gin"
"github.com/imroc/req/v3"
"github.com/stretchr/testify/require"
"github.com/tidwall/gjson"
)
@ -398,11 +400,38 @@ func TestCollectOpenAIImagePointers_RecognizesDirectAssets(t *testing.T) {
func TestResolveOpenAIImageBytes_PrefersInlineBase64(t *testing.T) {
data, err := resolveOpenAIImageBytes(context.Background(), nil, nil, "", openAIImagePointerInfo{
B64JSON: "data:image/png;base64,QUJD",
})
}, openAIUpstreamErrorBodyReadLimit)
require.NoError(t, err)
require.Equal(t, []byte("ABC"), data)
}
func TestNewOpenAIImageStatusError_UsesProvidedReadLimit(t *testing.T) {
padding := strings.Repeat("x", int(openAIUpstreamErrorBodyReadLimit)+1024)
body := fmt.Sprintf(`{"error":{"padding":"%s","message":"diagnostic-marker"}}`, padding)
resp := &req.Response{Response: &http.Response{
StatusCode: http.StatusBadGateway,
Header: http.Header{},
Body: io.NopCloser(strings.NewReader(body)),
}}
err := newOpenAIImageStatusError(resp, "download image bytes failed", int64(len(body)))
require.Error(t, err)
require.Equal(t, "diagnostic-marker", err.Error())
var statusErr *openAIImageStatusError
require.ErrorAs(t, err, &statusErr)
require.Len(t, statusErr.ResponseBody, len(body))
}
func TestOpenAIUpstreamErrorBodyReadLimitForConfig_RespectsDiagnosticLimit(t *testing.T) {
cfg := &config.Config{Gateway: config.GatewayConfig{
LogUpstreamErrorBody: true,
LogUpstreamErrorBodyMaxBytes: int(openAIUpstreamErrorBodyReadLimit) + 1024,
}}
require.Equal(t, int64(cfg.Gateway.LogUpstreamErrorBodyMaxBytes), openAIUpstreamErrorBodyReadLimitForConfig(cfg))
}
func TestAccountSupportsOpenAIImageCapability_OAuthSupportsNative(t *testing.T) {
account := &Account{
Platform: PlatformOpenAI,

6
backend/internal/service/openai_tool_continuation.go
View File

@ -157,7 +157,11 @@ func AnalyzeToolContinuationSignals(reqBody map[string]any) ToolContinuationSign
// ValidateFunctionCallOutputContextBytes 基于 raw JSON 校验工具输出续链，避免 handler 预校验阶段全量解码大 input。
func ValidateFunctionCallOutputContextBytes(body []byte) FunctionCallOutputValidation {
result := FunctionCallOutputValidation{}
input := gjson.GetBytes(body, "input")
if len(body) == 0 {
return result
}
// handler 热路径只读扫描 input，避免 GetBytes 为大 Responses body 复制整段 JSON。
input := parseRawJSONView(body).Get("input")
if !input.IsArray() {
return result
}