github/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-07 12:57:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

优化 Grok 传输超时与 Realtime 握手

Browse Source
This commit is contained in:
IanShaw
2026-08-20 01:02:49 -07:00
parent 953028718d
commit 5ade094318
6 changed files with 85 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/config/config.go
+7
View File
@@ -934,6 +934,9 @@ type GatewayConfig struct {
// OpenAIResponseHeaderTimeout: OpenAI/Codex 上游等待响应头的超时时间（秒），0表示无超时
// OpenAI/Codex 请求可能在上游排队较久；默认不使用通用响应头超时截断。
OpenAIResponseHeaderTimeout int `mapstructure:"openai_response_header_timeout"`
// GrokResponseHeaderTimeout bounds the pre-first-byte wait for xAI/Grok.
// A zero value uses the provider-safe default instead of the generic gateway timeout.
GrokResponseHeaderTimeout int `mapstructure:"grok_response_header_timeout"`
// OpenAIFirstOutputTimeoutSeconds: native HTTP Responses 首个语义输出超时（秒），0表示禁用。
OpenAIFirstOutputTimeoutSeconds int `mapstructure:"openai_first_output_timeout_seconds"`
// OpenAIHighEffortFirstOutputTimeoutSeconds: high/xhigh/max 推理的首个语义输出超时（秒）。
@@ -2327,6 +2330,7 @@ func setDefaults() {
// Gateway
viper.SetDefault("gateway.response_header_timeout", 600) // 600秒(10分钟)等待上游响应头，LLM高负载时可能排队较久
viper.SetDefault("gateway.openai_response_header_timeout", 0)
viper.SetDefault("gateway.grok_response_header_timeout", 120)
viper.SetDefault("gateway.openai_first_output_timeout_seconds", 0)
viper.SetDefault("gateway.openai_high_effort_first_output_timeout_seconds", 0)
viper.SetDefault("gateway.log_upstream_error_body", true)
@@ -3241,6 +3245,9 @@ func (c *Config) Validate() error {
if c.Gateway.OpenAIResponseHeaderTimeout < 0 {
return fmt.Errorf("gateway.openai_response_header_timeout must be non-negative")
}
if c.Gateway.GrokResponseHeaderTimeout < 0 || c.Gateway.GrokResponseHeaderTimeout > 1800 {
return fmt.Errorf("gateway.grok_response_header_timeout must be between 0-1800 seconds")
}
if c.Gateway.OpenAIFirstOutputTimeoutSeconds < 0 || c.Gateway.OpenAIFirstOutputTimeoutSeconds > 600 ||
(c.Gateway.OpenAIFirstOutputTimeoutSeconds > 0 && c.Gateway.OpenAIFirstOutputTimeoutSeconds < 30) {
return fmt.Errorf("gateway.openai_first_output_timeout_seconds must be 0 or between 30-600 seconds")
backend/internal/handler/grok_audio.go
+14 -4
View File
@@ -77,6 +77,20 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) {
h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Grok credential unavailable")
return
}
model := c.Query("model")
if strings.TrimSpace(model) == "" {
model = "grok-voice-latest"
}
// Complete the upstream handshake before sending HTTP 101. This keeps
// pre-accept failures representable as normal JSON errors to the client.
probeCtx, cancelProbe := context.WithTimeout(c.Request.Context(), 15*time.Second)
probeErr := h.gatewayService.ProbeGrokRealtime(probeCtx, selection.Account, token, model)
cancelProbe()
if probeErr != nil {
reqLog.Warn("grok_realtime.pre_accept_failed", zap.Error(probeErr))
h.errorResponse(c, http.StatusBadGateway, "upstream_error", "Grok realtime upstream unavailable")
return
}
conn, err := coderws.Accept(c.Writer, c.Request, &coderws.AcceptOptions{CompressionMode: coderws.CompressionContextTakeover})
if err != nil {
@@ -84,10 +98,6 @@ func (h *OpenAIGatewayHandler) GrokRealtime(c *gin.Context) {
}
defer func() { _ = conn.CloseNow() }()
model := c.Query("model")
if strings.TrimSpace(model) == "" {
model = "grok-voice-latest"
}
started := time.Now()
audioObserved, proxyErr := h.gatewayService.ProxyGrokRealtime(c.Request.Context(), c, conn, selection.Account, token, model)
elapsed := time.Since(started)
backend/internal/repository/http_upstream.go
+14 -6
View File
@@ -899,12 +899,20 @@ func (s *httpUpstreamService) resolvePoolSettings(isolation string, accountConcu
}
func (s *httpUpstreamService) applyProfilePoolSettings(settings poolSettings, profile service.HTTPUpstreamProfile) poolSettings {
if profile != service.HTTPUpstreamProfileOpenAI {
return settings
}
settings.responseHeaderTimeout = 0
if s != nil && s.cfg != nil && s.cfg.Gateway.OpenAIResponseHeaderTimeout > 0 {
settings.responseHeaderTimeout = time.Duration(s.cfg.Gateway.OpenAIResponseHeaderTimeout) * time.Second
switch profile {
case service.HTTPUpstreamProfileOpenAI:
settings.responseHeaderTimeout = 0
if s != nil && s.cfg != nil && s.cfg.Gateway.OpenAIResponseHeaderTimeout > 0 {
settings.responseHeaderTimeout = time.Duration(s.cfg.Gateway.OpenAIResponseHeaderTimeout) * time.Second
}
case service.HTTPUpstreamProfileGrok:
// Grok can stall before its first byte under capacity pressure. Keep the
// generic 600s gateway timeout from turning one request into a 10-minute
// resource hold; streaming after headers is unaffected.
settings.responseHeaderTimeout = 120 * time.Second
if s != nil && s.cfg != nil && s.cfg.Gateway.GrokResponseHeaderTimeout > 0 {
settings.responseHeaderTimeout = time.Duration(s.cfg.Gateway.GrokResponseHeaderTimeout) * time.Second
}
}
return settings
}
backend/internal/service/grok_audio.go
+40
View File
@@ -63,6 +63,7 @@ func (s *OpenAIGatewayService) ForwardGrokVoice(ctx context.Context, c *gin.Cont
}
upstreamCtx, release := detachUpstreamContext(ctx)
defer release()
upstreamCtx = WithHTTPUpstreamProfile(upstreamCtx, HTTPUpstreamProfileGrok)
method := http.MethodPost
if c != nil && c.Request != nil && strings.TrimSpace(c.Request.Method) != "" {
method = c.Request.Method
@@ -207,6 +208,45 @@ func (s *OpenAIGatewayService) ProxyGrokRealtime(ctx context.Context, c *gin.Con
return awaitGrokRealtimeAudioObserved(errCh, &audioObserved)
}
// ProbeGrokRealtime performs the upstream WebSocket handshake without sending
// any client-visible events. Handlers use it before accepting the downstream
// upgrade so authentication and endpoint failures remain ordinary HTTP errors.
func (s *OpenAIGatewayService) ProbeGrokRealtime(ctx context.Context, account *Account, token, model string) error {
if s == nil || account == nil {
return fmt.Errorf("realtime service and account are required")
}
if account.Platform != PlatformGrok {
return fmt.Errorf("account platform %s is not supported for grok realtime", account.Platform)
}
base, err := buildGrokVoiceURL(account, s.cfg, "realtime")
if err != nil {
return err
}
u, err := url.Parse(base)
if err != nil {
return err
}
u.Scheme = "wss"
q := u.Query()
q.Set("model", firstNonEmpty(model, "grok-voice-latest"))
u.RawQuery = q.Encode()
headers := http.Header{"Authorization": []string{"Bearer " + token}}
if account.IsGrokOAuth() && isGrokCLIProxyTarget(u.String()) {
applyGrokCLIHeaders(headers)
}
account.ApplyHeaderOverrides(headers)
proxyURL := ""
if account.ProxyID != nil && account.Proxy != nil {
proxyURL = account.Proxy.URL()
}
dialer := s.getOpenAIWSPassthroughDialer()
conn, _, _, err := dialer.Dial(ctx, u.String(), headers, proxyURL)
if err != nil {
return err
}
return conn.Close()
}
func awaitGrokRealtimeAudioObserved(errCh <-chan error, audioObserved *atomic.Bool) (bool, error) {
err := <-errCh
if audioObserved == nil {
backend/internal/service/grok_media.go
+8 -4
View File
@@ -658,6 +658,7 @@ func (s *OpenAIGatewayService) ForwardGrokMedia(
}
upstreamCtx, releaseUpstreamCtx := detachUpstreamContext(ctx)
defer releaseUpstreamCtx()
upstreamCtx = WithHTTPUpstreamProfile(upstreamCtx, HTTPUpstreamProfileGrok)
upstreamReq, err := http.NewRequestWithContext(upstreamCtx, endpoint.httpMethod(), targetURL, bodyReader)
if err != nil {
return nil, err
@@ -1288,11 +1289,14 @@ func (s *OpenAIGatewayService) handleGrokMediaErrorResponse(
Detail: upstreamDetail,
})
if kind == "failover" {
retryable, retryDelay, retryDeadline := grokSameAccountRetryMetadata(account, resp.StatusCode, body)
return nil, &UpstreamFailoverError{
StatusCode: resp.StatusCode,
ResponseBody: body,
ResponseHeaders: resp.Header.Clone(),
RetryableOnSameAccount: account.IsPoolMode() && account.IsPoolModeRetryableStatus(resp.StatusCode),
StatusCode: resp.StatusCode,
ResponseBody: body,
ResponseHeaders: resp.Header.Clone(),
RetryableOnSameAccount: retryable,
SameAccountRetryDelay: retryDelay,
SameAccountRetryDeadline: retryDeadline,
}
}
backend/internal/service/http_upstream_profile.go
+2 -1
View File
@@ -9,6 +9,7 @@ type HTTPUpstreamProfile string
const (
HTTPUpstreamProfileDefault HTTPUpstreamProfile = ""
HTTPUpstreamProfileOpenAI HTTPUpstreamProfile = "openai"
HTTPUpstreamProfileGrok HTTPUpstreamProfile = "grok"
)
type httpUpstreamProfileContextKey struct{}
@@ -35,7 +36,7 @@ func HTTPUpstreamProfileFromContext(ctx context.Context) HTTPUpstreamProfile {
return HTTPUpstreamProfileDefault
}
switch profile {
case HTTPUpstreamProfileOpenAI:
case HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileGrok:
return profile
default:
return HTTPUpstreamProfileDefault