Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Sign in / Register
Toggle navigation
Menu
Open sidebar
陈曦
sub2api
Commits
2d83941a
Commit
2d83941a
authored
Jan 09, 2026
by
shaw
Browse files
feat(antigravity): 添加 URL fallback 机制 (sandbox → daily → prod)
parent
afcfbb45
Changes
3
Hide whitespace changes
Inline
Side-by-side
backend/internal/pkg/antigravity/client.go
View file @
2d83941a
...
@@ -5,8 +5,11 @@ import (
...
@@ -5,8 +5,11 @@ import (
"bytes"
"bytes"
"context"
"context"
"encoding/json"
"encoding/json"
"errors"
"fmt"
"fmt"
"io"
"io"
"log"
"net"
"net/http"
"net/http"
"net/url"
"net/url"
"strings"
"strings"
...
@@ -22,10 +25,10 @@ func resolveHost(urlStr string) string {
...
@@ -22,10 +25,10 @@ func resolveHost(urlStr string) string {
return
parsed
.
Host
return
parsed
.
Host
}
}
// NewAPIRequest 创建 Antigravity API 请求(v1internal 端点)
// NewAPIRequest
WithURL 使用指定的 base URL
创建 Antigravity API 请求(v1internal 端点)
func
NewAPIRequest
(
ctx
context
.
Context
,
action
,
accessToken
string
,
body
[]
byte
)
(
*
http
.
Request
,
error
)
{
func
NewAPIRequest
WithURL
(
ctx
context
.
Context
,
baseURL
,
action
,
accessToken
string
,
body
[]
byte
)
(
*
http
.
Request
,
error
)
{
// 构建 URL,流式请求添加 ?alt=sse 参数
// 构建 URL,流式请求添加 ?alt=sse 参数
apiURL
:=
fmt
.
Sprintf
(
"%s/v1internal:%s"
,
B
aseURL
,
action
)
apiURL
:=
fmt
.
Sprintf
(
"%s/v1internal:%s"
,
b
aseURL
,
action
)
isStream
:=
action
==
"streamGenerateContent"
isStream
:=
action
==
"streamGenerateContent"
if
isStream
{
if
isStream
{
apiURL
+=
"?alt=sse"
apiURL
+=
"?alt=sse"
...
@@ -53,11 +56,15 @@ func NewAPIRequest(ctx context.Context, action, accessToken string, body []byte)
...
@@ -53,11 +56,15 @@ func NewAPIRequest(ctx context.Context, action, accessToken string, body []byte)
req
.
Host
=
host
req
.
Host
=
host
}
}
// 注意:requestType 已在 JSON body 的 V1InternalRequest 中设置,不需要 HTTP Header
return
req
,
nil
return
req
,
nil
}
}
// NewAPIRequest 使用默认 URL 创建 Antigravity API 请求(v1internal 端点)
// 向后兼容:仅使用默认 BaseURL
func
NewAPIRequest
(
ctx
context
.
Context
,
action
,
accessToken
string
,
body
[]
byte
)
(
*
http
.
Request
,
error
)
{
return
NewAPIRequestWithURL
(
ctx
,
BaseURL
,
action
,
accessToken
,
body
)
}
// TokenResponse Google OAuth token 响应
// TokenResponse Google OAuth token 响应
type
TokenResponse
struct
{
type
TokenResponse
struct
{
AccessToken
string
`json:"access_token"`
AccessToken
string
`json:"access_token"`
...
@@ -164,6 +171,38 @@ func NewClient(proxyURL string) *Client {
...
@@ -164,6 +171,38 @@ func NewClient(proxyURL string) *Client {
}
}
}
}
// isConnectionError 判断是否为连接错误(网络超时、DNS 失败、连接拒绝)
func
isConnectionError
(
err
error
)
bool
{
if
err
==
nil
{
return
false
}
// 检查超时错误
var
netErr
net
.
Error
if
errors
.
As
(
err
,
&
netErr
)
&&
netErr
.
Timeout
()
{
return
true
}
// 检查连接错误(DNS 失败、连接拒绝)
var
opErr
*
net
.
OpError
if
errors
.
As
(
err
,
&
opErr
)
{
return
true
}
// 检查 URL 错误
var
urlErr
*
url
.
Error
return
errors
.
As
(
err
,
&
urlErr
)
}
// shouldFallbackToNextURL 判断是否应切换到下一个 URL
// 仅连接错误和 HTTP 429 触发 URL 降级
func
shouldFallbackToNextURL
(
err
error
,
statusCode
int
)
bool
{
if
isConnectionError
(
err
)
{
return
true
}
return
statusCode
==
http
.
StatusTooManyRequests
}
// ExchangeCode 用 authorization code 交换 token
// ExchangeCode 用 authorization code 交换 token
func
(
c
*
Client
)
ExchangeCode
(
ctx
context
.
Context
,
code
,
codeVerifier
string
)
(
*
TokenResponse
,
error
)
{
func
(
c
*
Client
)
ExchangeCode
(
ctx
context
.
Context
,
code
,
codeVerifier
string
)
(
*
TokenResponse
,
error
)
{
params
:=
url
.
Values
{}
params
:=
url
.
Values
{}
...
@@ -272,6 +311,7 @@ func (c *Client) GetUserInfo(ctx context.Context, accessToken string) (*UserInfo
...
@@ -272,6 +311,7 @@ func (c *Client) GetUserInfo(ctx context.Context, accessToken string) (*UserInfo
}
}
// LoadCodeAssist 获取账户信息,返回解析后的结构体和原始 JSON
// LoadCodeAssist 获取账户信息,返回解析后的结构体和原始 JSON
// 支持 URL fallback:sandbox → daily → prod
func
(
c
*
Client
)
LoadCodeAssist
(
ctx
context
.
Context
,
accessToken
string
)
(
*
LoadCodeAssistResponse
,
map
[
string
]
any
,
error
)
{
func
(
c
*
Client
)
LoadCodeAssist
(
ctx
context
.
Context
,
accessToken
string
)
(
*
LoadCodeAssistResponse
,
map
[
string
]
any
,
error
)
{
reqBody
:=
LoadCodeAssistRequest
{}
reqBody
:=
LoadCodeAssistRequest
{}
reqBody
.
Metadata
.
IDEType
=
"ANTIGRAVITY"
reqBody
.
Metadata
.
IDEType
=
"ANTIGRAVITY"
...
@@ -281,40 +321,65 @@ func (c *Client) LoadCodeAssist(ctx context.Context, accessToken string) (*LoadC
...
@@ -281,40 +321,65 @@ func (c *Client) LoadCodeAssist(ctx context.Context, accessToken string) (*LoadC
return
nil
,
nil
,
fmt
.
Errorf
(
"序列化请求失败: %w"
,
err
)
return
nil
,
nil
,
fmt
.
Errorf
(
"序列化请求失败: %w"
,
err
)
}
}
url
:=
BaseURL
+
"/v1internal:loadCodeAssist"
// 获取可用的 URL 列表
req
,
err
:=
http
.
NewRequestWithContext
(
ctx
,
http
.
MethodPost
,
url
,
strings
.
NewReader
(
string
(
bodyBytes
))
)
availableURLs
:=
DefaultURLAvailability
.
GetAvailableURLs
(
)
if
err
!=
nil
{
if
len
(
availableURLs
)
==
0
{
return
nil
,
nil
,
fmt
.
Errorf
(
"创建请求失败: %w"
,
err
)
availableURLs
=
BaseURLs
// 所有 URL 都不可用时,重试所有
}
}
req
.
Header
.
Set
(
"Authorization"
,
"Bearer "
+
accessToken
)
req
.
Header
.
Set
(
"Content-Type"
,
"application/json"
)
req
.
Header
.
Set
(
"User-Agent"
,
UserAgent
)
resp
,
err
:=
c
.
httpClient
.
Do
(
req
)
var
lastErr
error
if
err
!=
nil
{
for
urlIdx
,
baseURL
:=
range
availableURLs
{
return
nil
,
nil
,
fmt
.
Errorf
(
"loadCodeAssist 请求失败: %w"
,
err
)
apiURL
:=
baseURL
+
"/v1internal:loadCodeAssist"
}
req
,
err
:=
http
.
NewRequestWithContext
(
ctx
,
http
.
MethodPost
,
apiURL
,
strings
.
NewReader
(
string
(
bodyBytes
)))
defer
func
()
{
_
=
resp
.
Body
.
Close
()
}()
if
err
!=
nil
{
lastErr
=
fmt
.
Errorf
(
"创建请求失败: %w"
,
err
)
continue
}
req
.
Header
.
Set
(
"Authorization"
,
"Bearer "
+
accessToken
)
req
.
Header
.
Set
(
"Content-Type"
,
"application/json"
)
req
.
Header
.
Set
(
"User-Agent"
,
UserAgent
)
resp
,
err
:=
c
.
httpClient
.
Do
(
req
)
if
err
!=
nil
{
lastErr
=
fmt
.
Errorf
(
"loadCodeAssist 请求失败: %w"
,
err
)
if
shouldFallbackToNextURL
(
err
,
0
)
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"[antigravity] loadCodeAssist URL fallback: %s -> %s"
,
baseURL
,
availableURLs
[
urlIdx
+
1
])
continue
}
return
nil
,
nil
,
lastErr
}
respBodyBytes
,
err
:=
io
.
ReadAll
(
resp
.
Body
)
respBodyBytes
,
err
:=
io
.
ReadAll
(
resp
.
Body
)
if
err
!=
nil
{
_
=
resp
.
Body
.
Close
()
// 立即关闭,避免循环内 defer 导致的资源泄漏
return
nil
,
nil
,
fmt
.
Errorf
(
"读取响应失败: %w"
,
err
)
if
err
!=
nil
{
}
return
nil
,
nil
,
fmt
.
Errorf
(
"读取响应失败: %w"
,
err
)
}
if
resp
.
StatusCode
!=
http
.
StatusOK
{
// 检查是否需要 URL 降级
return
nil
,
nil
,
fmt
.
Errorf
(
"loadCodeAssist 失败 (HTTP %d): %s"
,
resp
.
StatusCode
,
string
(
respBodyBytes
))
if
shouldFallbackToNextURL
(
nil
,
resp
.
StatusCode
)
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
}
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"[antigravity] loadCodeAssist URL fallback (HTTP %d): %s -> %s"
,
resp
.
StatusCode
,
baseURL
,
availableURLs
[
urlIdx
+
1
])
continue
}
var
loadResp
LoadCodeAssistResponse
if
resp
.
StatusCode
!=
http
.
StatusOK
{
if
err
:=
json
.
Unmarshal
(
respBodyBytes
,
&
loadResp
);
err
!=
nil
{
return
nil
,
nil
,
fmt
.
Errorf
(
"loadCodeAssist 失败 (HTTP %d): %s"
,
resp
.
StatusCode
,
string
(
respBodyBytes
))
return
nil
,
nil
,
fmt
.
Errorf
(
"响应解析失败: %w"
,
err
)
}
}
// 解析原始 JSON 为 map
var
loadResp
LoadCodeAssistResponse
var
rawResp
map
[
string
]
any
if
err
:=
json
.
Unmarshal
(
respBodyBytes
,
&
loadResp
);
err
!=
nil
{
_
=
json
.
Unmarshal
(
respBodyBytes
,
&
rawResp
)
return
nil
,
nil
,
fmt
.
Errorf
(
"响应解析失败: %w"
,
err
)
}
// 解析原始 JSON 为 map
var
rawResp
map
[
string
]
any
_
=
json
.
Unmarshal
(
respBodyBytes
,
&
rawResp
)
return
&
loadResp
,
rawResp
,
nil
}
return
&
loadResp
,
rawResp
,
nil
return
nil
,
nil
,
lastErr
}
}
// ModelQuotaInfo 模型配额信息
// ModelQuotaInfo 模型配额信息
...
@@ -339,6 +404,7 @@ type FetchAvailableModelsResponse struct {
...
@@ -339,6 +404,7 @@ type FetchAvailableModelsResponse struct {
}
}
// FetchAvailableModels 获取可用模型和配额信息,返回解析后的结构体和原始 JSON
// FetchAvailableModels 获取可用模型和配额信息,返回解析后的结构体和原始 JSON
// 支持 URL fallback:sandbox → daily → prod
func
(
c
*
Client
)
FetchAvailableModels
(
ctx
context
.
Context
,
accessToken
,
projectID
string
)
(
*
FetchAvailableModelsResponse
,
map
[
string
]
any
,
error
)
{
func
(
c
*
Client
)
FetchAvailableModels
(
ctx
context
.
Context
,
accessToken
,
projectID
string
)
(
*
FetchAvailableModelsResponse
,
map
[
string
]
any
,
error
)
{
reqBody
:=
FetchAvailableModelsRequest
{
Project
:
projectID
}
reqBody
:=
FetchAvailableModelsRequest
{
Project
:
projectID
}
bodyBytes
,
err
:=
json
.
Marshal
(
reqBody
)
bodyBytes
,
err
:=
json
.
Marshal
(
reqBody
)
...
@@ -346,38 +412,63 @@ func (c *Client) FetchAvailableModels(ctx context.Context, accessToken, projectI
...
@@ -346,38 +412,63 @@ func (c *Client) FetchAvailableModels(ctx context.Context, accessToken, projectI
return
nil
,
nil
,
fmt
.
Errorf
(
"序列化请求失败: %w"
,
err
)
return
nil
,
nil
,
fmt
.
Errorf
(
"序列化请求失败: %w"
,
err
)
}
}
apiURL
:=
BaseURL
+
"/v1internal:fetchAvailableModels"
// 获取可用的 URL 列表
req
,
err
:=
http
.
NewRequestWithContext
(
ctx
,
http
.
MethodPost
,
apiURL
,
strings
.
NewReader
(
string
(
bodyBytes
))
)
availableURLs
:=
DefaultURLAvailability
.
GetAvailableURLs
(
)
if
err
!=
nil
{
if
len
(
availableURLs
)
==
0
{
return
nil
,
nil
,
fmt
.
Errorf
(
"创建请求失败: %w"
,
err
)
availableURLs
=
BaseURLs
// 所有 URL 都不可用时,重试所有
}
}
req
.
Header
.
Set
(
"Authorization"
,
"Bearer "
+
accessToken
)
req
.
Header
.
Set
(
"Content-Type"
,
"application/json"
)
req
.
Header
.
Set
(
"User-Agent"
,
UserAgent
)
resp
,
err
:=
c
.
httpClient
.
Do
(
req
)
var
lastErr
error
if
err
!=
nil
{
for
urlIdx
,
baseURL
:=
range
availableURLs
{
return
nil
,
nil
,
fmt
.
Errorf
(
"fetchAvailableModels 请求失败: %w"
,
err
)
apiURL
:=
baseURL
+
"/v1internal:fetchAvailableModels"
}
req
,
err
:=
http
.
NewRequestWithContext
(
ctx
,
http
.
MethodPost
,
apiURL
,
strings
.
NewReader
(
string
(
bodyBytes
)))
defer
func
()
{
_
=
resp
.
Body
.
Close
()
}()
if
err
!=
nil
{
lastErr
=
fmt
.
Errorf
(
"创建请求失败: %w"
,
err
)
continue
}
req
.
Header
.
Set
(
"Authorization"
,
"Bearer "
+
accessToken
)
req
.
Header
.
Set
(
"Content-Type"
,
"application/json"
)
req
.
Header
.
Set
(
"User-Agent"
,
UserAgent
)
resp
,
err
:=
c
.
httpClient
.
Do
(
req
)
if
err
!=
nil
{
lastErr
=
fmt
.
Errorf
(
"fetchAvailableModels 请求失败: %w"
,
err
)
if
shouldFallbackToNextURL
(
err
,
0
)
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"[antigravity] fetchAvailableModels URL fallback: %s -> %s"
,
baseURL
,
availableURLs
[
urlIdx
+
1
])
continue
}
return
nil
,
nil
,
lastErr
}
respBodyBytes
,
err
:=
io
.
ReadAll
(
resp
.
Body
)
respBodyBytes
,
err
:=
io
.
ReadAll
(
resp
.
Body
)
if
err
!=
nil
{
_
=
resp
.
Body
.
Close
()
// 立即关闭,避免循环内 defer 导致的资源泄漏
return
nil
,
nil
,
fmt
.
Errorf
(
"读取响应失败: %w"
,
err
)
if
err
!=
nil
{
}
return
nil
,
nil
,
fmt
.
Errorf
(
"读取响应失败: %w"
,
err
)
}
if
resp
.
StatusCode
!=
http
.
StatusOK
{
// 检查是否需要 URL 降级
return
nil
,
nil
,
fmt
.
Errorf
(
"fetchAvailableModels 失败 (HTTP %d): %s"
,
resp
.
StatusCode
,
string
(
respBodyBytes
))
if
shouldFallbackToNextURL
(
nil
,
resp
.
StatusCode
)
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
}
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"[antigravity] fetchAvailableModels URL fallback (HTTP %d): %s -> %s"
,
resp
.
StatusCode
,
baseURL
,
availableURLs
[
urlIdx
+
1
])
continue
}
var
modelsResp
FetchAvailableModelsResponse
if
resp
.
StatusCode
!=
http
.
StatusOK
{
if
err
:=
json
.
Unmarshal
(
respBodyBytes
,
&
modelsResp
);
err
!=
nil
{
return
nil
,
nil
,
fmt
.
Errorf
(
"fetchAvailableModels 失败 (HTTP %d): %s"
,
resp
.
StatusCode
,
string
(
respBodyBytes
))
return
nil
,
nil
,
fmt
.
Errorf
(
"响应解析失败: %w"
,
err
)
}
}
// 解析原始 JSON 为 map
var
modelsResp
FetchAvailableModelsResponse
var
rawResp
map
[
string
]
any
if
err
:=
json
.
Unmarshal
(
respBodyBytes
,
&
modelsResp
);
err
!=
nil
{
_
=
json
.
Unmarshal
(
respBodyBytes
,
&
rawResp
)
return
nil
,
nil
,
fmt
.
Errorf
(
"响应解析失败: %w"
,
err
)
}
// 解析原始 JSON 为 map
var
rawResp
map
[
string
]
any
_
=
json
.
Unmarshal
(
respBodyBytes
,
&
rawResp
)
return
&
modelsResp
,
rawResp
,
nil
}
return
&
modelsResp
,
rawResp
,
nil
return
nil
,
nil
,
lastErr
}
}
backend/internal/pkg/antigravity/oauth.go
View file @
2d83941a
...
@@ -32,17 +32,79 @@ const (
...
@@ -32,17 +32,79 @@ const (
"https://www.googleapis.com/auth/cclog "
+
"https://www.googleapis.com/auth/cclog "
+
"https://www.googleapis.com/auth/experimentsandconfigs"
"https://www.googleapis.com/auth/experimentsandconfigs"
// API 端点
// 优先使用 sandbox daily URL,配额更宽松
BaseURL
=
"https://daily-cloudcode-pa.sandbox.googleapis.com"
// User-Agent(模拟官方客户端)
// User-Agent(模拟官方客户端)
UserAgent
=
"antigravity/1.104.0 darwin/arm64"
UserAgent
=
"antigravity/1.104.0 darwin/arm64"
// Session 过期时间
// Session 过期时间
SessionTTL
=
30
*
time
.
Minute
SessionTTL
=
30
*
time
.
Minute
// URL 可用性 TTL(不可用 URL 的恢复时间)
URLAvailabilityTTL
=
5
*
time
.
Minute
)
)
// BaseURLs 定义 Antigravity API 端点,按优先级排序
// fallback 顺序: sandbox → daily → prod
var
BaseURLs
=
[]
string
{
"https://daily-cloudcode-pa.sandbox.googleapis.com"
,
// sandbox
"https://daily-cloudcode-pa.googleapis.com"
,
// daily
"https://cloudcode-pa.googleapis.com"
,
// prod
}
// BaseURL 默认 URL(保持向后兼容)
var
BaseURL
=
BaseURLs
[
0
]
// URLAvailability 管理 URL 可用性状态(带 TTL 自动恢复)
type
URLAvailability
struct
{
mu
sync
.
RWMutex
unavailable
map
[
string
]
time
.
Time
// URL -> 恢复时间
ttl
time
.
Duration
}
// DefaultURLAvailability 全局 URL 可用性管理器
var
DefaultURLAvailability
=
NewURLAvailability
(
URLAvailabilityTTL
)
// NewURLAvailability 创建 URL 可用性管理器
func
NewURLAvailability
(
ttl
time
.
Duration
)
*
URLAvailability
{
return
&
URLAvailability
{
unavailable
:
make
(
map
[
string
]
time
.
Time
),
ttl
:
ttl
,
}
}
// MarkUnavailable 标记 URL 临时不可用
func
(
u
*
URLAvailability
)
MarkUnavailable
(
url
string
)
{
u
.
mu
.
Lock
()
defer
u
.
mu
.
Unlock
()
u
.
unavailable
[
url
]
=
time
.
Now
()
.
Add
(
u
.
ttl
)
}
// IsAvailable 检查 URL 是否可用
func
(
u
*
URLAvailability
)
IsAvailable
(
url
string
)
bool
{
u
.
mu
.
RLock
()
defer
u
.
mu
.
RUnlock
()
expiry
,
exists
:=
u
.
unavailable
[
url
]
if
!
exists
{
return
true
}
return
time
.
Now
()
.
After
(
expiry
)
}
// GetAvailableURLs 返回可用的 URL 列表(保持优先级顺序)
func
(
u
*
URLAvailability
)
GetAvailableURLs
()
[]
string
{
u
.
mu
.
RLock
()
defer
u
.
mu
.
RUnlock
()
now
:=
time
.
Now
()
result
:=
make
([]
string
,
0
,
len
(
BaseURLs
))
for
_
,
url
:=
range
BaseURLs
{
expiry
,
exists
:=
u
.
unavailable
[
url
]
if
!
exists
||
now
.
After
(
expiry
)
{
result
=
append
(
result
,
url
)
}
}
return
result
}
// OAuthSession 保存 OAuth 授权流程的临时状态
// OAuthSession 保存 OAuth 授权流程的临时状态
type
OAuthSession
struct
{
type
OAuthSession
struct
{
State
string
`json:"state"`
State
string
`json:"state"`
...
...
backend/internal/service/antigravity_gateway_service.go
View file @
2d83941a
...
@@ -10,6 +10,7 @@ import (
...
@@ -10,6 +10,7 @@ import (
"io"
"io"
"log"
"log"
mathrand
"math/rand"
mathrand
"math/rand"
"net"
"net/http"
"net/http"
"strings"
"strings"
"sync/atomic"
"sync/atomic"
...
@@ -27,6 +28,32 @@ const (
...
@@ -27,6 +28,32 @@ const (
antigravityRetryMaxDelay
=
16
*
time
.
Second
antigravityRetryMaxDelay
=
16
*
time
.
Second
)
)
// isAntigravityConnectionError 判断是否为连接错误(网络超时、DNS 失败、连接拒绝)
func
isAntigravityConnectionError
(
err
error
)
bool
{
if
err
==
nil
{
return
false
}
// 检查超时错误
var
netErr
net
.
Error
if
errors
.
As
(
err
,
&
netErr
)
&&
netErr
.
Timeout
()
{
return
true
}
// 检查连接错误(DNS 失败、连接拒绝)
var
opErr
*
net
.
OpError
return
errors
.
As
(
err
,
&
opErr
)
}
// shouldAntigravityFallbackToNextURL 判断是否应切换到下一个 URL
// 仅连接错误和 HTTP 429 触发 URL 降级
func
shouldAntigravityFallbackToNextURL
(
err
error
,
statusCode
int
)
bool
{
if
isAntigravityConnectionError
(
err
)
{
return
true
}
return
statusCode
==
http
.
StatusTooManyRequests
}
// getSessionID 从 gin.Context 获取 session_id(用于日志追踪)
// getSessionID 从 gin.Context 获取 session_id(用于日志追踪)
func
getSessionID
(
c
*
gin
.
Context
)
string
{
func
getSessionID
(
c
*
gin
.
Context
)
string
{
if
c
==
nil
{
if
c
==
nil
{
...
@@ -181,45 +208,70 @@ func (s *AntigravityGatewayService) TestConnection(ctx context.Context, account
...
@@ -181,45 +208,70 @@ func (s *AntigravityGatewayService) TestConnection(ctx context.Context, account
return
nil
,
fmt
.
Errorf
(
"构建请求失败: %w"
,
err
)
return
nil
,
fmt
.
Errorf
(
"构建请求失败: %w"
,
err
)
}
}
// 构建 HTTP 请求(总是使用流式 endpoint,与官方客户端一致)
req
,
err
:=
antigravity
.
NewAPIRequest
(
ctx
,
"streamGenerateContent"
,
accessToken
,
requestBody
)
if
err
!=
nil
{
return
nil
,
err
}
// 调试日志:Test 请求信息
log
.
Printf
(
"[antigravity-Test] account=%s request_size=%d url=%s"
,
account
.
Name
,
len
(
requestBody
),
req
.
URL
.
String
())
// 代理 URL
// 代理 URL
proxyURL
:=
""
proxyURL
:=
""
if
account
.
ProxyID
!=
nil
&&
account
.
Proxy
!=
nil
{
if
account
.
ProxyID
!=
nil
&&
account
.
Proxy
!=
nil
{
proxyURL
=
account
.
Proxy
.
URL
()
proxyURL
=
account
.
Proxy
.
URL
()
}
}
//
发送请求
//
URL fallback 循环
resp
,
err
:=
s
.
httpUpstream
.
Do
(
req
,
proxyURL
,
account
.
ID
,
account
.
Concurrency
)
availableURLs
:=
antigravity
.
DefaultURLAvailability
.
GetAvailableURLs
(
)
if
err
!=
nil
{
if
len
(
availableURLs
)
==
0
{
return
nil
,
fmt
.
Errorf
(
"请求失败: %w"
,
err
)
availableURLs
=
antigravity
.
BaseURLs
// 所有 URL 都不可用时,重试所有
}
}
defer
func
()
{
_
=
resp
.
Body
.
Close
()
}()
// 读取响应
var
lastErr
error
respBody
,
err
:=
io
.
ReadAll
(
io
.
LimitReader
(
resp
.
Body
,
2
<<
20
))
for
urlIdx
,
baseURL
:=
range
availableURLs
{
if
err
!=
nil
{
// 构建 HTTP 请求(总是使用流式 endpoint,与官方客户端一致)
return
nil
,
fmt
.
Errorf
(
"读取响应失败: %w"
,
err
)
req
,
err
:=
antigravity
.
NewAPIRequestWithURL
(
ctx
,
baseURL
,
"streamGenerateContent"
,
accessToken
,
requestBody
)
}
if
err
!=
nil
{
lastErr
=
err
continue
}
if
resp
.
StatusCode
>=
400
{
// 调试日志:Test 请求信息
return
nil
,
fmt
.
Errorf
(
"API 返回 %d: %s"
,
resp
.
StatusCode
,
string
(
respBody
))
log
.
Printf
(
"[antigravity-Test] account=%s request_size=%d url=%s"
,
account
.
Name
,
len
(
requestBody
),
req
.
URL
.
String
())
}
// 解析流式响应,提取文本
// 发送请求
text
:=
extractTextFromSSEResponse
(
respBody
)
resp
,
err
:=
s
.
httpUpstream
.
Do
(
req
,
proxyURL
,
account
.
ID
,
account
.
Concurrency
)
if
err
!=
nil
{
lastErr
=
fmt
.
Errorf
(
"请求失败: %w"
,
err
)
if
shouldAntigravityFallbackToNextURL
(
err
,
0
)
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
antigravity
.
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"[antigravity-Test] URL fallback: %s -> %s"
,
baseURL
,
availableURLs
[
urlIdx
+
1
])
continue
}
return
nil
,
lastErr
}
return
&
TestConnectionResult
{
// 读取响应
Text
:
text
,
respBody
,
err
:=
io
.
ReadAll
(
io
.
LimitReader
(
resp
.
Body
,
2
<<
20
))
MappedModel
:
mappedModel
,
_
=
resp
.
Body
.
Close
()
// 立即关闭,避免循环内 defer 导致的资源泄漏
},
nil
if
err
!=
nil
{
return
nil
,
fmt
.
Errorf
(
"读取响应失败: %w"
,
err
)
}
// 检查是否需要 URL 降级
if
shouldAntigravityFallbackToNextURL
(
nil
,
resp
.
StatusCode
)
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
antigravity
.
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"[antigravity-Test] URL fallback (HTTP %d): %s -> %s"
,
resp
.
StatusCode
,
baseURL
,
availableURLs
[
urlIdx
+
1
])
continue
}
if
resp
.
StatusCode
>=
400
{
return
nil
,
fmt
.
Errorf
(
"API 返回 %d: %s"
,
resp
.
StatusCode
,
string
(
respBody
))
}
// 解析流式响应,提取文本
text
:=
extractTextFromSSEResponse
(
respBody
)
return
&
TestConnectionResult
{
Text
:
text
,
MappedModel
:
mappedModel
,
},
nil
}
return
nil
,
lastErr
}
}
// buildGeminiTestRequest 构建 Gemini 格式测试请求
// buildGeminiTestRequest 构建 Gemini 格式测试请求
...
@@ -484,62 +536,86 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
...
@@ -484,62 +536,86 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
// 如果客户端请求非流式,在响应处理阶段会收集完整流式响应后转换返回
// 如果客户端请求非流式,在响应处理阶段会收集完整流式响应后转换返回
action
:=
"streamGenerateContent"
action
:=
"streamGenerateContent"
// URL fallback 循环
availableURLs
:=
antigravity
.
DefaultURLAvailability
.
GetAvailableURLs
()
if
len
(
availableURLs
)
==
0
{
availableURLs
=
antigravity
.
BaseURLs
// 所有 URL 都不可用时,重试所有
}
// 重试循环
// 重试循环
var
resp
*
http
.
Response
var
resp
*
http
.
Response
for
attempt
:=
1
;
attempt
<=
antigravityMaxRetries
;
attempt
++
{
urlFallbackLoop
:
// 检查 context 是否已取消(客户端断开连接)
for
urlIdx
,
baseURL
:=
range
availableURLs
{
select
{
for
attempt
:=
1
;
attempt
<=
antigravityMaxRetries
;
attempt
++
{
case
<-
ctx
.
Done
()
:
// 检查 context 是否已取消(客户端断开连接)
log
.
Printf
(
"%s status=context_canceled error=%v"
,
prefix
,
ctx
.
Err
())
select
{
return
nil
,
ctx
.
Err
()
case
<-
ctx
.
Done
()
:
default
:
log
.
Printf
(
"%s status=context_canceled error=%v"
,
prefix
,
ctx
.
Err
())
}
return
nil
,
ctx
.
Err
()
default
:
}
upstreamReq
,
err
:=
antigravity
.
NewAPIRequest
(
ctx
,
action
,
accessToken
,
geminiBody
)
upstreamReq
,
err
:=
antigravity
.
NewAPIRequest
WithURL
(
ctx
,
baseURL
,
action
,
accessToken
,
geminiBody
)
if
err
!=
nil
{
if
err
!=
nil
{
return
nil
,
err
return
nil
,
err
}
}
resp
,
err
=
s
.
httpUpstream
.
Do
(
upstreamReq
,
proxyURL
,
account
.
ID
,
account
.
Concurrency
)
resp
,
err
=
s
.
httpUpstream
.
Do
(
upstreamReq
,
proxyURL
,
account
.
ID
,
account
.
Concurrency
)
if
err
!=
nil
{
if
err
!=
nil
{
if
attempt
<
antigravityMaxRetries
{
// 检查是否应触发 URL 降级
log
.
Printf
(
"%s status=request_failed retry=%d/%d error=%v"
,
prefix
,
attempt
,
antigravityMaxRetries
,
err
)
if
shouldAntigravityFallbackToNextURL
(
err
,
0
)
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
if
!
sleepAntigravityBackoffWithContext
(
ctx
,
attempt
)
{
antigravity
.
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"%s
status=context_canceled_during_backoff"
,
prefix
)
log
.
Printf
(
"%s
URL fallback (connection error): %s -> %s"
,
prefix
,
baseURL
,
availableURLs
[
urlIdx
+
1
]
)
return
nil
,
ctx
.
Err
()
continue
urlFallbackLoop
}
}
continue
if
attempt
<
antigravityMaxRetries
{
log
.
Printf
(
"%s status=request_failed retry=%d/%d error=%v"
,
prefix
,
attempt
,
antigravityMaxRetries
,
err
)
if
!
sleepAntigravityBackoffWithContext
(
ctx
,
attempt
)
{
log
.
Printf
(
"%s status=context_canceled_during_backoff"
,
prefix
)
return
nil
,
ctx
.
Err
()
}
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"
)
}
}
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"
)
}
if
resp
.
StatusCode
>=
400
&&
s
.
shouldRetryUpstreamError
(
resp
.
StatusCode
)
{
// 检查是否应触发 URL 降级(仅 429)
respBody
,
_
:=
io
.
ReadAll
(
io
.
LimitReader
(
resp
.
Body
,
2
<<
20
))
if
resp
.
StatusCode
==
http
.
StatusTooManyRequests
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
_
=
resp
.
Body
.
Close
()
respBody
,
_
:=
io
.
ReadAll
(
io
.
LimitReader
(
resp
.
Body
,
2
<<
20
))
_
=
resp
.
Body
.
Close
()
antigravity
.
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"%s URL fallback (HTTP 429): %s -> %s body=%s"
,
prefix
,
baseURL
,
availableURLs
[
urlIdx
+
1
],
truncateForLog
(
respBody
,
200
))
continue
urlFallbackLoop
}
if
resp
.
StatusCode
>=
400
&&
s
.
shouldRetryUpstreamError
(
resp
.
StatusCode
)
{
respBody
,
_
:=
io
.
ReadAll
(
io
.
LimitReader
(
resp
.
Body
,
2
<<
20
))
_
=
resp
.
Body
.
Close
()
if
attempt
<
antigravityMaxRetries
{
if
attempt
<
antigravityMaxRetries
{
log
.
Printf
(
"%s status=%d retry=%d/%d body=%s"
,
prefix
,
resp
.
StatusCode
,
attempt
,
antigravityMaxRetries
,
truncateForLog
(
respBody
,
500
))
log
.
Printf
(
"%s status=%d retry=%d/%d body=%s"
,
prefix
,
resp
.
StatusCode
,
attempt
,
antigravityMaxRetries
,
truncateForLog
(
respBody
,
500
))
if
!
sleepAntigravityBackoffWithContext
(
ctx
,
attempt
)
{
if
!
sleepAntigravityBackoffWithContext
(
ctx
,
attempt
)
{
log
.
Printf
(
"%s status=context_canceled_during_backoff"
,
prefix
)
log
.
Printf
(
"%s status=context_canceled_during_backoff"
,
prefix
)
return
nil
,
ctx
.
Err
()
return
nil
,
ctx
.
Err
()
}
continue
}
}
continue
// 所有重试都失败,标记限流状态
}
if
resp
.
StatusCode
==
429
{
// 所有重试都失败,标记限流状态
s
.
handleUpstreamError
(
ctx
,
prefix
,
account
,
resp
.
StatusCode
,
resp
.
Header
,
respBody
)
if
resp
.
StatusCode
==
429
{
}
s
.
handleUpstreamError
(
ctx
,
prefix
,
account
,
resp
.
StatusCode
,
resp
.
Header
,
respBody
)
// 最后一次尝试也失败
}
resp
=
&
http
.
Response
{
// 最后一次尝试也失败
StatusCode
:
resp
.
StatusCode
,
resp
=
&
http
.
Response
{
Header
:
resp
.
Header
.
Clone
(),
StatusC
od
e
:
resp
.
StatusCode
,
B
od
y
:
io
.
NopCloser
(
bytes
.
NewReader
(
respBody
))
,
Header
:
resp
.
Header
.
Clone
(),
}
Body
:
io
.
NopCloser
(
bytes
.
NewReader
(
respBody
)),
break
urlFallbackLoop
}
}
break
}
break
break
urlFallbackLoop
}
}
}
defer
func
()
{
_
=
resp
.
Body
.
Close
()
}()
defer
func
()
{
_
=
resp
.
Body
.
Close
()
}()
...
@@ -1003,61 +1079,85 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
...
@@ -1003,61 +1079,85 @@ func (s *AntigravityGatewayService) ForwardGemini(ctx context.Context, c *gin.Co
// 如果客户端请求非流式,在响应处理阶段会收集完整流式响应后返回
// 如果客户端请求非流式,在响应处理阶段会收集完整流式响应后返回
upstreamAction
:=
"streamGenerateContent"
upstreamAction
:=
"streamGenerateContent"
// URL fallback 循环
availableURLs
:=
antigravity
.
DefaultURLAvailability
.
GetAvailableURLs
()
if
len
(
availableURLs
)
==
0
{
availableURLs
=
antigravity
.
BaseURLs
// 所有 URL 都不可用时,重试所有
}
// 重试循环
// 重试循环
var
resp
*
http
.
Response
var
resp
*
http
.
Response
for
attempt
:=
1
;
attempt
<=
antigravityMaxRetries
;
attempt
++
{
urlFallbackLoop
:
// 检查 context 是否已取消(客户端断开连接)
for
urlIdx
,
baseURL
:=
range
availableURLs
{
select
{
for
attempt
:=
1
;
attempt
<=
antigravityMaxRetries
;
attempt
++
{
case
<-
ctx
.
Done
()
:
// 检查 context 是否已取消(客户端断开连接)
log
.
Printf
(
"%s status=context_canceled error=%v"
,
prefix
,
ctx
.
Err
())
select
{
return
nil
,
ctx
.
Err
()
case
<-
ctx
.
Done
()
:
default
:
log
.
Printf
(
"%s status=context_canceled error=%v"
,
prefix
,
ctx
.
Err
())
}
return
nil
,
ctx
.
Err
()
default
:
}
upstreamReq
,
err
:=
antigravity
.
NewAPIRequest
(
ctx
,
upstreamAction
,
accessToken
,
wrappedBody
)
upstreamReq
,
err
:=
antigravity
.
NewAPIRequest
WithURL
(
ctx
,
baseURL
,
upstreamAction
,
accessToken
,
wrappedBody
)
if
err
!=
nil
{
if
err
!=
nil
{
return
nil
,
err
return
nil
,
err
}
}
resp
,
err
=
s
.
httpUpstream
.
Do
(
upstreamReq
,
proxyURL
,
account
.
ID
,
account
.
Concurrency
)
resp
,
err
=
s
.
httpUpstream
.
Do
(
upstreamReq
,
proxyURL
,
account
.
ID
,
account
.
Concurrency
)
if
err
!=
nil
{
if
err
!=
nil
{
if
attempt
<
antigravityMaxRetries
{
// 检查是否应触发 URL 降级
log
.
Printf
(
"%s status=request_failed retry=%d/%d error=%v"
,
prefix
,
attempt
,
antigravityMaxRetries
,
err
)
if
shouldAntigravityFallbackToNextURL
(
err
,
0
)
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
if
!
sleepAntigravityBackoffWithContext
(
ctx
,
attempt
)
{
antigravity
.
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"%s
status=context_canceled_during_backoff"
,
prefix
)
log
.
Printf
(
"%s
URL fallback (connection error): %s -> %s"
,
prefix
,
baseURL
,
availableURLs
[
urlIdx
+
1
]
)
return
nil
,
ctx
.
Err
()
continue
urlFallbackLoop
}
}
continue
if
attempt
<
antigravityMaxRetries
{
log
.
Printf
(
"%s status=request_failed retry=%d/%d error=%v"
,
prefix
,
attempt
,
antigravityMaxRetries
,
err
)
if
!
sleepAntigravityBackoffWithContext
(
ctx
,
attempt
)
{
log
.
Printf
(
"%s status=context_canceled_during_backoff"
,
prefix
)
return
nil
,
ctx
.
Err
()
}
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"
)
}
}
log
.
Printf
(
"%s status=request_failed retries_exhausted error=%v"
,
prefix
,
err
)
return
nil
,
s
.
writeGoogleError
(
c
,
http
.
StatusBadGateway
,
"Upstream request failed after retries"
)
}
if
resp
.
StatusCode
>=
400
&&
s
.
shouldRetryUpstreamError
(
resp
.
StatusCode
)
{
// 检查是否应触发 URL 降级(仅 429)
respBody
,
_
:=
io
.
ReadAll
(
io
.
LimitReader
(
resp
.
Body
,
2
<<
20
))
if
resp
.
StatusCode
==
http
.
StatusTooManyRequests
&&
urlIdx
<
len
(
availableURLs
)
-
1
{
_
=
resp
.
Body
.
Close
()
respBody
,
_
:=
io
.
ReadAll
(
io
.
LimitReader
(
resp
.
Body
,
2
<<
20
))
_
=
resp
.
Body
.
Close
()
antigravity
.
DefaultURLAvailability
.
MarkUnavailable
(
baseURL
)
log
.
Printf
(
"%s URL fallback (HTTP 429): %s -> %s body=%s"
,
prefix
,
baseURL
,
availableURLs
[
urlIdx
+
1
],
truncateForLog
(
respBody
,
200
))
continue
urlFallbackLoop
}
if
resp
.
StatusCode
>=
400
&&
s
.
shouldRetryUpstreamError
(
resp
.
StatusCode
)
{
respBody
,
_
:=
io
.
ReadAll
(
io
.
LimitReader
(
resp
.
Body
,
2
<<
20
))
_
=
resp
.
Body
.
Close
()
if
attempt
<
antigravityMaxRetries
{
if
attempt
<
antigravityMaxRetries
{
log
.
Printf
(
"%s status=%d retry=%d/%d"
,
prefix
,
resp
.
StatusCode
,
attempt
,
antigravityMaxRetries
)
log
.
Printf
(
"%s status=%d retry=%d/%d"
,
prefix
,
resp
.
StatusCode
,
attempt
,
antigravityMaxRetries
)
if
!
sleepAntigravityBackoffWithContext
(
ctx
,
attempt
)
{
if
!
sleepAntigravityBackoffWithContext
(
ctx
,
attempt
)
{
log
.
Printf
(
"%s status=context_canceled_during_backoff"
,
prefix
)
log
.
Printf
(
"%s status=context_canceled_during_backoff"
,
prefix
)
return
nil
,
ctx
.
Err
()
return
nil
,
ctx
.
Err
()
}
continue
}
}
continue
// 所有重试都失败,标记限流状态
}
if
resp
.
StatusCode
==
429
{
// 所有重试都失败,标记限流状态
s
.
handleUpstreamError
(
ctx
,
prefix
,
account
,
resp
.
StatusCode
,
resp
.
Header
,
respBody
)
if
resp
.
StatusCode
==
429
{
}
s
.
handleUpstreamError
(
ctx
,
prefix
,
account
,
resp
.
StatusCode
,
resp
.
Header
,
respBody
)
resp
=
&
http
.
Response
{
}
StatusCode
:
resp
.
StatusCode
,
resp
=
&
http
.
Response
{
Header
:
resp
.
Header
.
Clone
(),
StatusC
od
e
:
resp
.
StatusCode
,
B
od
y
:
io
.
NopCloser
(
bytes
.
NewReader
(
respBody
))
,
Header
:
resp
.
Header
.
Clone
(),
}
Body
:
io
.
NopCloser
(
bytes
.
NewReader
(
respBody
)),
break
urlFallbackLoop
}
}
break
}
break
break
urlFallbackLoop
}
}
}
defer
func
()
{
defer
func
()
{
if
resp
!=
nil
&&
resp
.
Body
!=
nil
{
if
resp
!=
nil
&&
resp
.
Body
!=
nil
{
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment