oadmin/sub2api
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'slovx2/main'

Browse Source
This commit is contained in:
shaw
2026-01-04 19:51:17 +08:00
parent a527559526 63453fbfa0
commit 4df712624e
7 changed files with 240 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/internal/handler/gateway_handler.go
View File

@@ -373,7 +373,7 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
continue
}
// 错误响应已在Forward中处理，这里只记录日志
log.Printf("Forward request failed: %v", err)
log.Printf("Account %d: Forward request failed: %v", account.ID, err)
return
}

2
backend/internal/handler/openai_gateway_handler.go
View File

@@ -225,7 +225,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
continue
}
// Error response already handled in Forward, just log
log.Printf("Forward request failed: %v", err)
log.Printf("Account %d: Forward request failed: %v", account.ID, err)
return
}

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

@@ -20,11 +20,27 @@ import (
const (
antigravityStickySessionTTL = time.Hour
antigravityMaxRetries = 5
antigravityMaxRetries = 3
antigravityRetryBaseDelay = 1 * time.Second
antigravityRetryMaxDelay = 16 * time.Second
)
// getSessionID 从 gin.Context 获取 session_id（用于日志追踪）
func getSessionID(c *gin.Context) string {
if c == nil {
return ""
}
return c.GetHeader("session_id")
}
// logPrefix 生成统一的日志前缀
func logPrefix(sessionID, accountName string) string {
if sessionID != "" {
return fmt.Sprintf("[antigravity-Forward] session=%s account=%s", sessionID, accountName)
}
return fmt.Sprintf("[antigravity-Forward] account=%s", accountName)
}
// Antigravity 直接支持的模型（精确匹配透传）
var antigravitySupportedModels = map[string]bool{
"claude-opus-4-5-thinking": true,
@@ -46,10 +62,11 @@ var antigravityPrefixMapping = []struct {
target string
}{
// 长前缀优先
{"gemini-3-pro-image", "gemini-3-pro-image"}, // gemini-3-pro-image-preview 等
{"claude-3-5-sonnet", "claude-sonnet-4-5"}, // 旧版 claude-3-5-sonnet-xxx
{"claude-sonnet-4-5", "claude-sonnet-4-5"}, // claude-sonnet-4-5-xxx
{"claude-haiku-4-5", "claude-sonnet-4-5"}, // claude-haiku-4-5-xxx → sonnet
{"gemini-2.5-flash-image", "gemini-3-pro-image"}, // gemini-2.5-flash-image → 3-pro-image
{"gemini-3-pro-image", "gemini-3-pro-image"}, // gemini-3-pro-image-preview 等
{"claude-3-5-sonnet", "claude-sonnet-4-5"}, // 旧版 claude-3-5-sonnet-xxx
{"claude-sonnet-4-5", "claude-sonnet-4-5"}, // claude-sonnet-4-5-xxx
{"claude-haiku-4-5", "claude-sonnet-4-5"}, // claude-haiku-4-5-xxx → sonnet
{"claude-opus-4-5", "claude-opus-4-5-thinking"},
{"claude-3-haiku", "claude-sonnet-4-5"}, // 旧版 claude-3-haiku-xxx → sonnet
{"claude-sonnet-4", "claude-sonnet-4-5"},
@@ -310,6 +327,8 @@ func (s *AntigravityGatewayService) unwrapV1InternalResponse(body []byte) ([]byt
// Forward 转发 Claude 协议请求（Claude → Gemini 转换）
func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context, account *Account, body []byte) (*ForwardResult, error) {
startTime := time.Now()
sessionID := getSessionID(c)
prefix := logPrefix(sessionID, account.Name)
// 解析 Claude 请求
var claudeReq antigravity.ClaudeRequest
@@ -364,10 +383,11 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
resp, err = s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
if err != nil {
if attempt < antigravityMaxRetries {
log.Printf("Antigravity account %d: upstream request failed, retry %d/%d: %v", account.ID, attempt, antigravityMaxRetries, err)
log.Printf("%s status=request_failed retry=%d/%d error=%v", prefix, attempt, antigravityMaxRetries, err)
sleepAntigravityBackoff(attempt)
continue
}
log.Printf("%s status=request_failed retries_exhausted error=%v", prefix, err)
return nil, s.writeClaudeError(c, http.StatusBadGateway, "upstream_error", "Upstream request failed after retries")
}
@@ -376,13 +396,13 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
_ = resp.Body.Close()
if attempt < antigravityMaxRetries {
log.Printf("Antigravity account %d: upstream status %d, retry %d/%d", account.ID, resp.StatusCode, attempt, antigravityMaxRetries)
log.Printf("%s status=%d retry=%d/%d", prefix, resp.StatusCode, attempt, antigravityMaxRetries)
sleepAntigravityBackoff(attempt)
continue
}
// 所有重试都失败，标记限流状态
if resp.StatusCode == 429 {
s.handleUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
s.handleUpstreamError(ctx, prefix, account, resp.StatusCode, resp.Header, respBody)
}
// 最后一次尝试也失败
resp = &http.Response{
@@ -400,7 +420,7 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
// 处理错误响应
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
s.handleUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
s.handleUpstreamError(ctx, prefix, account, resp.StatusCode, resp.Header, respBody)
if s.shouldFailoverUpstreamError(resp.StatusCode) {
return nil, &UpstreamFailoverError{StatusCode: resp.StatusCode}
@@ -419,6 +439,7 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
if claudeReq.Stream {
streamRes, err := s.handleClaudeStreamingResponse(c, resp, startTime, originalModel)
if err != nil {
log.Printf("%s status=stream_error error=%v", prefix, err)
return nil, err
}
usage = streamRes.usage
@@ -443,6 +464,8 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
// ForwardGemini 转发 Gemini 协议请求
func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Context, account *Account, originalModel string, action string, stream bool, body []byte) (*ForwardResult, error) {
startTime := time.Now()
sessionID := getSessionID(c)
prefix := logPrefix(sessionID, account.Name)
if strings.TrimSpace(originalModel) == "" {
return nil, s.writeGoogleError(c, http.StatusBadRequest, "Missing model in URL")
@@ -518,10 +541,11 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
resp, err = s.httpUpstream.Do(upstreamReq, proxyURL, account.ID, account.Concurrency)
if err != nil {
if attempt < antigravityMaxRetries {
log.Printf("Antigravity account %d: upstream request failed, retry %d/%d: %v", account.ID, attempt, antigravityMaxRetries, err)
log.Printf("%s status=request_failed retry=%d/%d error=%v", prefix, attempt, antigravityMaxRetries, err)
sleepAntigravityBackoff(attempt)
continue
}
log.Printf("%s status=request_failed retries_exhausted error=%v", prefix, err)
return nil, s.writeGoogleError(c, http.StatusBadGateway, "Upstream request failed after retries")
}
@@ -530,13 +554,13 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
_ = resp.Body.Close()
if attempt < antigravityMaxRetries {
log.Printf("Antigravity account %d: upstream status %d, retry %d/%d", account.ID, resp.StatusCode, attempt, antigravityMaxRetries)
log.Printf("%s status=%d retry=%d/%d", prefix, resp.StatusCode, attempt, antigravityMaxRetries)
sleepAntigravityBackoff(attempt)
continue
}
// 所有重试都失败，标记限流状态
if resp.StatusCode == 429 {
s.handleUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
s.handleUpstreamError(ctx, prefix, account, resp.StatusCode, resp.Header, respBody)
}
resp = &http.Response{
StatusCode: resp.StatusCode,
@@ -558,7 +582,7 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
// 处理错误响应
if resp.StatusCode >= 400 {
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
s.handleUpstreamError(ctx, account, resp.StatusCode, resp.Header, respBody)
s.handleUpstreamError(ctx, prefix, account, resp.StatusCode, resp.Header, respBody)
if s.shouldFailoverUpstreamError(resp.StatusCode) {
return nil, &UpstreamFailoverError{StatusCode: resp.StatusCode}
@@ -580,6 +604,7 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
if stream || upstreamAction == "streamGenerateContent" {
streamRes, err := s.handleGeminiStreamingResponse(c, resp, startTime)
if err != nil {
log.Printf("%s status=stream_error error=%v", prefix, err)
return nil, err
}
usage = streamRes.usage
@@ -628,7 +653,7 @@ func sleepAntigravityBackoff(attempt int) {
sleepGeminiBackoff(attempt) // 复用 Gemini 的退避逻辑
}
func (s *AntigravityGatewayService) handleUpstreamError(ctx context.Context, account *Account, statusCode int, headers http.Header, body []byte) {
func (s *AntigravityGatewayService) handleUpstreamError(ctx context.Context, prefix string, account *Account, statusCode int, headers http.Header, body []byte) {
// 429 使用 Gemini 格式解析（从 body 解析重置时间）
if statusCode == 429 {
resetAt := ParseGeminiRateLimitResetTime(body)
@@ -639,17 +664,23 @@ func (s *AntigravityGatewayService) handleUpstreamError(ctx context.Context, acc
defaultDur = 5 * time.Minute
}
ra := time.Now().Add(defaultDur)
log.Printf("%s status=429 rate_limited reset_in=%v (fallback)", prefix, defaultDur)
_ = s.accountRepo.SetRateLimited(ctx, account.ID, ra)
return
}
_ = s.accountRepo.SetRateLimited(ctx, account.ID, time.Unix(*resetAt, 0))
resetTime := time.Unix(*resetAt, 0)
log.Printf("%s status=429 rate_limited reset_at=%v reset_in=%v", prefix, resetTime.Format("15:04:05"), time.Until(resetTime).Truncate(time.Second))
_ = s.accountRepo.SetRateLimited(ctx, account.ID, resetTime)
return
}
// 其他错误码继续使用 rateLimitService
if s.rateLimitService == nil {
return
}
s.rateLimitService.HandleUpstreamError(ctx, account, statusCode, headers, body)
shouldDisable := s.rateLimitService.HandleUpstreamError(ctx, account, statusCode, headers, body)
if shouldDisable {
log.Printf("%s status=%d marked_error", prefix, statusCode)
}
}
type antigravityStreamResult struct {
@@ -758,7 +789,7 @@ func (s *AntigravityGatewayService) writeClaudeError(c *gin.Context, status int,
func (s *AntigravityGatewayService) writeMappedClaudeError(c *gin.Context, upstreamStatus int, body []byte) error {
// 记录上游错误详情便于调试
log.Printf("Antigravity upstream error %d: %s", upstreamStatus, string(body))
log.Printf("[antigravity-Forward] upstream_error status=%d body=%s", upstreamStatus, string(body))
var statusCode int
var errType, errMsg string
@@ -832,7 +863,7 @@ func (s *AntigravityGatewayService) handleClaudeNonStreamingResponse(c *gin.Cont
// 转换 Gemini 响应为 Claude 格式
claudeResp, agUsage, err := antigravity.TransformGeminiToClaude(body, originalModel)
if err != nil {
log.Printf("Transform Gemini to Claude failed: %v, body: %s", err, string(body))
log.Printf("[antigravity-Forward] transform_error error=%v body=%s", err, string(body))
return nil, s.writeClaudeError(c, http.StatusBadGateway, "upstream_error", "Failed to parse upstream response")
}

44
backend/internal/service/token_refresh_service.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log"
"strings"
"sync"
"time"
@@ -171,6 +172,15 @@ func (s *TokenRefreshService) refreshWithRetry(ctx context.Context, account *Acc
return nil
}
// Antigravity 账户：不可重试错误直接标记 error 状态并返回
if account.Platform == PlatformAntigravity && isNonRetryableRefreshError(err) {
errorMsg := fmt.Sprintf("Token refresh failed (non-retryable): %v", err)
if setErr := s.accountRepo.SetError(ctx, account.ID, errorMsg); setErr != nil {
log.Printf("[TokenRefresh] Failed to set error status for account %d: %v", account.ID, setErr)
}
return err
}
lastErr = err
log.Printf("[TokenRefresh] Account %d attempt %d/%d failed: %v",
account.ID, attempt, s.cfg.MaxRetries, err)
@@ -183,11 +193,37 @@ func (s *TokenRefreshService) refreshWithRetry(ctx context.Context, account *Acc
}
}
// 所有重试都失败，标记账号为error状态
errorMsg := fmt.Sprintf("Token refresh failed after %d retries: %v", s.cfg.MaxRetries, lastErr)
if err := s.accountRepo.SetError(ctx, account.ID, errorMsg); err != nil {
log.Printf("[TokenRefresh] Failed to set error status for account %d: %v", account.ID, err)
// Antigravity 账户：其他错误仅记录日志，不标记 error（可能是临时网络问题）
// 其他平台账户：重试失败后标记 error
if account.Platform == PlatformAntigravity {
log.Printf("[TokenRefresh] Account %d: refresh failed after %d retries: %v", account.ID, s.cfg.MaxRetries, lastErr)
} else {
errorMsg := fmt.Sprintf("Token refresh failed after %d retries: %v", s.cfg.MaxRetries, lastErr)
if err := s.accountRepo.SetError(ctx, account.ID, errorMsg); err != nil {
log.Printf("[TokenRefresh] Failed to set error status for account %d: %v", account.ID, err)
}
}
return lastErr
}
// isNonRetryableRefreshError 判断是否为不可重试的刷新错误
// 这些错误通常表示凭证已失效，需要用户重新授权
func isNonRetryableRefreshError(err error) bool {
if err == nil {
return false
}
msg := strings.ToLower(err.Error())
nonRetryable := []string{
"invalid_grant", // refresh_token 已失效
"invalid_client", // 客户端配置错误
"unauthorized_client", // 客户端未授权
"access_denied", // 访问被拒绝
}
for _, needle := range nonRetryable {
if strings.Contains(msg, needle) {
return true
}
}
return false
}