BCAI/easyai-ai-gateway
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(provider): 延长媒体超时并终止超时轮转

Browse Source 
将图像和视频的默认 HTTP/轮询超时分别提高到 20 分钟和 30 分钟。媒体任务超时后直接失败，不再重试、切换客户端或标记为 upstream_submission_unknown。OpenAI 图像端点遇到上游明确拒绝 response_format 时自动移除并缓存兼容结论。

验证：go test ./... -count=1；go vet ./...；gofmt；相关 Shell bash -n、ShellCheck 与发布脚本回归测试。
This commit is contained in:
wangbo
2026-08-05 13:58:02 +08:00
parent 9ce9053d7b
commit dd10a807df
23 changed files with 336 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/internal/runner/proxy.go
+28 -11
View File
@@ -5,7 +5,6 @@ import (
"net/url"
"strings"
"sync"
"time"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/netproxy"
@@ -19,8 +18,6 @@ type httpClientCache struct {
custom map[string]*http.Client
}
const providerHTTPClientTimeout = 10 * time.Minute
const (
providerHTTPMaxIdleConnections = 2048
providerHTTPMaxIdleConnectionsPerHost = 1024
@@ -34,16 +31,19 @@ func newHTTPClientCache() *httpClientCache {
}
}
func (s *Service) httpClientForCandidate(candidate store.RuntimeModelCandidate, simulated bool) (*http.Client, error) {
func (s *Service) httpClientForCandidate(candidate store.RuntimeModelCandidate, simulated bool, kind string) (*http.Client, error) {
var client *http.Client
if simulated {
return s.httpClients.none, nil
client = s.httpClients.none
return providerHTTPClientForKind(client, kind), nil
}
// Some Worker sites have direct provider egress but are not authorized to
// use a platform's shared proxy. Keep this override explicitly scoped to
// platform UUIDs so one site's routing exception cannot bypass proxies for
// unrelated providers or platforms.
if platformIDListed(s.cfg.PlatformProxyBypassIDs, candidate.PlatformID) {
return s.httpClients.none, nil
client = s.httpClients.none
return providerHTTPClientForKind(client, kind), nil
}
config, err := netproxy.Normalize(netproxy.FromPlatformConfig(candidate.PlatformConfig))
if err != nil {
@@ -52,14 +52,22 @@ func (s *Service) httpClientForCandidate(candidate store.RuntimeModelCandidate,
switch config.Mode {
case netproxy.ModeGlobal:
if strings.TrimSpace(s.cfg.GlobalHTTPProxy) != "" {
return s.httpClients.customClient(s.cfg.GlobalHTTPProxy)
client, err = s.httpClients.customClient(s.cfg.GlobalHTTPProxy)
if err != nil {
return nil, err
}
break
}
return s.httpClients.global, nil
client = s.httpClients.global
case netproxy.ModeCustom:
return s.httpClients.customClient(config.HTTPProxy)
client, err = s.httpClients.customClient(config.HTTPProxy)
if err != nil {
return nil, err
}
default:
return s.httpClients.none, nil
client = s.httpClients.none
}
return providerHTTPClientForKind(client, kind), nil
}
func platformIDListed(raw string, platformID string) bool {
@@ -100,7 +108,16 @@ func newHTTPClient(proxy func(*http.Request) (*url.URL, error)) *http.Client {
transport.MaxIdleConns = providerHTTPMaxIdleConnections
transport.MaxIdleConnsPerHost = providerHTTPMaxIdleConnectionsPerHost
return &http.Client{
Timeout: providerHTTPClientTimeout,
Timeout: clients.ProviderRequestTimeout(""),
Transport: transport,
}
}
func providerHTTPClientForKind(base *http.Client, kind string) *http.Client {
if base == nil || base.Timeout == clients.ProviderRequestTimeout(kind) {
return base
}
client := *base
client.Timeout = clients.ProviderRequestTimeout(kind)
return &client
}
apps/api/internal/runner/proxy_test.go
+17 -4
View File
@@ -16,6 +16,19 @@ func TestProviderHTTPClientTimeoutAllowsLongRunningMediaRequests(t *testing.T) {
if client.Timeout != 10*time.Minute {
t.Fatalf("unexpected provider HTTP timeout: got %s want %s", client.Timeout, 10*time.Minute)
}
for _, test := range []struct {
kind string
want time.Duration
}{
{kind: "images.generations", want: 20 * time.Minute},
{kind: "images.edits", want: 20 * time.Minute},
{kind: "videos.generations", want: 30 * time.Minute},
{kind: "chat.completions", want: 10 * time.Minute},
} {
if got := providerHTTPClientForKind(client, test.kind).Timeout; got != test.want {
t.Fatalf("provider HTTP timeout for %s: got %s want %s", test.kind, got, test.want)
}
}
transport, ok := client.Transport.(*http.Transport)
if !ok {
t.Fatalf("provider transport type = %T, want *http.Transport", client.Transport)
@@ -51,7 +64,7 @@ func TestPlatformProxyModeNoneIgnoresEnvironmentProxy(t *testing.T) {
client, err := testProxyService(config.Config{}).httpClientForCandidate(store.RuntimeModelCandidate{
PlatformConfig: map[string]any{"networkProxy": map[string]any{"mode": "none"}},
}, false)
}, false, "chat.completions")
if err != nil {
t.Fatalf("build http client: %v", err)
}
@@ -86,7 +99,7 @@ func TestPlatformProxyModeCustomUsesConfiguredHTTPProxy(t *testing.T) {
client, err := testProxyService(config.Config{}).httpClientForCandidate(store.RuntimeModelCandidate{
PlatformConfig: map[string]any{"networkProxy": map[string]any{"mode": "custom", "httpProxy": proxy.URL}},
}, false)
}, false, "chat.completions")
if err != nil {
t.Fatalf("build http client: %v", err)
}
@@ -121,7 +134,7 @@ func TestPlatformProxyBypassIDUsesDirectConnection(t *testing.T) {
}).httpClientForCandidate(store.RuntimeModelCandidate{
PlatformID: "official-gemini-platform",
PlatformConfig: map[string]any{"networkProxy": map[string]any{"mode": "custom", "httpProxy": proxy.URL}},
}, false)
}, false, "chat.completions")
if err != nil {
t.Fatalf("build bypassed http client: %v", err)
}
@@ -153,7 +166,7 @@ func TestPlatformProxyModeGlobalUsesConfiguredGlobalHTTPProxy(t *testing.T) {
client, err := testProxyService(config.Config{GlobalHTTPProxy: proxy.URL}).httpClientForCandidate(store.RuntimeModelCandidate{
PlatformConfig: map[string]any{"networkProxy": map[string]any{"mode": "global"}},
}, false)
}, false, "chat.completions")
if err != nil {
t.Fatalf("build http client: %v", err)
}
apps/api/internal/runner/retry_decision.go
+21
View File
@@ -63,6 +63,7 @@ type resolveCandidateFailureInput struct {
FailoverExpired bool
Async bool
DownstreamStarted bool
TaskKind string
}
type priorityDemoteDecision struct {
@@ -150,6 +151,20 @@ func failoverDecisionForCandidate(runnerPolicy store.RunnerPolicy, candidate sto
func resolveCandidateFailure(input resolveCandidateFailureInput) failureDecision {
info := failureInfoFromError(input.Err)
if terminalMediaTimeout(input.TaskKind, input.Err) {
return failureDecision{
Route: "stop",
Effect: "none",
Reason: "media_timeout_terminal",
Match: policyRuleMatch{
Source: "gateway_media_timeout",
Policy: "terminalTimeout",
Rule: "taskKind",
Value: strings.TrimSpace(input.TaskKind),
},
Info: info,
}
}
if isResultPersistenceFailure(input.Err) {
return failureDecision{
Route: "stop",
@@ -260,6 +275,12 @@ func resolveCandidateFailure(input resolveCandidateFailureInput) failureDecision
}
}
func terminalMediaTimeout(kind string, err error) bool {
kind = strings.TrimSpace(kind)
return (strings.HasPrefix(kind, "images.") || strings.HasPrefix(kind, "videos.")) &&
strings.EqualFold(strings.TrimSpace(clients.ErrorCode(err)), "timeout")
}
func failoverEffect(action string) string {
switch action {
case "cooldown_and_next":
apps/api/internal/runner/retry_decision_test.go
+19
View File
@@ -442,6 +442,25 @@ func TestResolveCandidateFailureStopsAfterDownstreamStarts(t *testing.T) {
}
}
func TestResolveCandidateFailureStopsTerminalMediaTimeout(t *testing.T) {
decision := resolveCandidateFailure(resolveCandidateFailureInput{
RunnerPolicy: store.RunnerPolicy{
Status: "active",
FailoverPolicy: map[string]any{"enabled": true},
},
Candidate: store.RuntimeModelCandidate{ModelRetryPolicy: map[string]any{"enabled": true, "maxAttempts": 3}},
Err: &clients.ClientError{Code: "timeout", Message: "upstream request timed out", Retryable: true},
ClientAttempt: 1,
MaxClientAttempts: 3,
HasNextCandidate: true,
TaskKind: "images.edits",
})
if decision.Route != "stop" || decision.Effect != "none" || decision.Reason != "media_timeout_terminal" {
t.Fatalf("media timeout must stop without retry or rotation, got %+v", decision)
}
}
func TestResolveCandidateFailureLegacyPolicyPrecedence(t *testing.T) {
candidate := store.RuntimeModelCandidate{
DegradePolicy: map[string]any{
apps/api/internal/runner/service.go
+8 -2
View File
@@ -81,6 +81,10 @@ type upstreamSubmissionUnknownError struct {
Cause error
}
func shouldClassifyUpstreamSubmissionUnknown(simulated bool, submissionStatus string, err error) bool {
return !simulated && submissionStatus == "submitting" && clients.ErrorCode(err) != "timeout"
}
func (e *upstreamSubmissionUnknownError) Error() string {
return "upstream submission result is unknown"
}
@@ -1105,6 +1109,7 @@ candidatesLoop:
FailoverExpired: failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration),
Async: task.AsyncMode,
DownstreamStarted: downstreamStarted.Load(),
TaskKind: task.Kind,
})
if candidateDecision.Route == "requeue" {
if _, platformLimited := platformModelRateLimitError(err); platformLimited {
@@ -1162,6 +1167,7 @@ candidatesLoop:
FailoverExpired: failoverTimeBudgetExceeded(executeStartedAt, maxFailoverDuration),
Async: task.AsyncMode,
DownstreamStarted: downstreamStarted.Load(),
TaskKind: task.Kind,
})
candidateDecisionAttempt = attemptNo
candidateClientAttempt = clientAttempt
@@ -1431,7 +1437,7 @@ func (s *Service) runCandidate(
}
defer s.store.RecordClientRelease(context.WithoutCancel(ctx), candidate.ClientID, "")
requestHTTPClient, err := s.httpClientForCandidate(candidate, simulated)
requestHTTPClient, err := s.httpClientForCandidate(candidate, simulated, task.Kind)
if err != nil {
_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
AttemptID: attemptID,
@@ -1685,7 +1691,7 @@ func (s *Service) runCandidate(
ErrorMessage: err.Error(),
})
_ = s.emit(ctx, task.ID, "task.attempt.failed", "running", "attempt_failed", 0.45, err.Error(), map[string]any{"attempt": attemptNo, "retryable": retryable, "requestId": requestID, "statusCode": clients.ErrorResponseMetadata(err).StatusCode, "metrics": metrics}, simulated)
if !simulated && submissionStatus == "submitting" {
if shouldClassifyUpstreamSubmissionUnknown(simulated, submissionStatus, err) {
return clients.Response{}, &upstreamSubmissionUnknownError{AttemptID: attemptID, Cause: err}
}
return clients.Response{}, err
apps/api/internal/runner/submission_timeout_test.go
+18
View File
@@ -0,0 +1,18 @@
package runner
import (
"testing"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
)
func TestTimeoutDoesNotBecomeUpstreamSubmissionUnknown(t *testing.T) {
timeoutErr := &clients.ClientError{Code: "timeout", Message: "upstream request timed out", Retryable: false}
if shouldClassifyUpstreamSubmissionUnknown(false, "submitting", timeoutErr) {
t.Fatal("a definitive provider timeout must remain timeout instead of manual-review unknown")
}
networkErr := &clients.ClientError{Code: "network", Message: "connection reset", Retryable: true}
if !shouldClassifyUpstreamSubmissionUnknown(false, "submitting", networkErr) {
t.Fatal("an ambiguous non-timeout disconnect must retain manual-review protection")
}
}
apps/api/internal/runner/task_cancel.go
+1 -1
View File
@@ -129,7 +129,7 @@ func (s *Service) CancelVolcesVideoTask(ctx context.Context, task store.GatewayT
if !found || !isVolcesCancellationCandidate(candidate) {
return local, nil
}
httpClient, err := s.httpClientForCandidate(candidate, false)
httpClient, err := s.httpClientForCandidate(candidate, false, "videos.generations")
if err != nil {
return TaskCancelResult{}, err
}
apps/api/internal/runner/voice_clone.go
+1 -1
View File
@@ -210,7 +210,7 @@ func (s *Service) DeleteClonedVoice(ctx context.Context, user *auth.User, rawID
if err != nil {
return DeletedClonedVoiceResult{}, err
}
requestHTTPClient, err := s.httpClientForCandidate(candidate, false)
requestHTTPClient, err := s.httpClientForCandidate(candidate, false, "voice.clone")
if err != nil {
return DeletedClonedVoiceResult{}, err
}