diff --git a/backend/internal/application/gateway/selector.go b/backend/internal/application/gateway/selector.go index db0fe95ca..c8d1d0a1a 100644 --- a/backend/internal/application/gateway/selector.go +++ b/backend/internal/application/gateway/selector.go @@ -1183,24 +1183,31 @@ func (s *Selector) markMissingThinking(ctx context.Context, credential account.C } func (s *Selector) MarkFailure(ctx context.Context, credential account.Credential, status int, retryAfter time.Duration) { - _ = s.markFailure(ctx, credential, credential.FailureCount, credential.FailureCount+1, status, retryAfter) + _ = s.markFailure(ctx, credential, credential.FailureCount, credential.FailureCount+1, status, retryAfter, status == 0) } // MarkFailureAfterSuccess records a stream failure from a fresh health baseline. // The upstream already returned a successful response header, so failures that // preceded this request must not be carried into the new cooldown calculation. func (s *Selector) MarkFailureAfterSuccess(ctx context.Context, credential account.Credential, status int, retryAfter time.Duration) error { - return s.markFailure(ctx, credential, 0, 1, status, retryAfter) + return s.markFailure(ctx, credential, 0, 1, status, retryAfter, status == 0) } -func (s *Selector) markFailure(ctx context.Context, credential account.Credential, baselineFailureCount, nextFailureCount, status int, retryAfter time.Duration) error { +// markSoftFailure preserves the real upstream status for diagnostics while +// applying the bounded, non-accumulating health penalty used for transient +// network and provider-wide failures. +func (s *Selector) markSoftFailure(ctx context.Context, credential account.Credential, status int, retryAfter time.Duration) error { + return s.markFailure(ctx, credential, credential.FailureCount, credential.FailureCount+1, status, retryAfter, true) +} + +func (s *Selector) markFailure(ctx context.Context, credential account.Credential, baselineFailureCount, nextFailureCount, status int, retryAfter time.Duration, soft bool) error { _, cooldownBase, cooldownMax, _ := s.routingConfig() - // 网络/超时（status 0）只短隔离本号，不累加失败次数，避免瞬时抖动把号池指数冻空。 - // 上游返回的 4xx/5xx 仍按原指数冷却：那是上游明确给出的状态，不是本地网络抖动。 - softNetwork := status == 0 + // Soft failures only isolate this account briefly and never accumulate the + // durable failure count. Hard account-scoped 4xx responses retain the + // exponential policy below. effectiveFailureCount := nextFailureCount cooldown := cooldownBase - if softNetwork { + if soft { effectiveFailureCount = baselineFailureCount cooldown = softNetworkCooldown if retryAfter > cooldown { diff --git a/backend/internal/application/gateway/selector_test.go b/backend/internal/application/gateway/selector_test.go index 7ea3ffba0..a68e589e8 100644 --- a/backend/internal/application/gateway/selector_test.go +++ b/backend/internal/application/gateway/selector_test.go @@ -1631,7 +1631,48 @@ func TestMarkFailureSoftNetworkCooldown(t *testing.T) { } before = time.Now().UTC() - selector.MarkFailure(ctx, hard, 0, 0) + if err := selector.markSoftFailure(ctx, hard, http.StatusGatewayTimeout, 0); err != nil { + t.Fatal(err) + } + soft5xx, err := accounts.Get(ctx, credential.ID) + if err != nil { + t.Fatal(err) + } + if soft5xx.FailureCount != hard.FailureCount { + t.Fatalf("soft 5xx failure count = %d, want preserved %d", soft5xx.FailureCount, hard.FailureCount) + } + if soft5xx.LastError != "upstream status 504" { + t.Fatalf("soft 5xx diagnostic = %q", soft5xx.LastError) + } + if soft5xx.CooldownUntil == nil { + t.Fatal("soft 5xx did not set cooldown") + } + cooldown = soft5xx.CooldownUntil.Sub(before) + if cooldown < 4*time.Second || cooldown > 6*time.Second { + t.Fatalf("soft 5xx cooldown = %s, want ~5s", cooldown) + } + + before = time.Now().UTC() + if err := selector.markSoftFailure(ctx, soft5xx, http.StatusServiceUnavailable, 12*time.Second); err != nil { + t.Fatal(err) + } + softRetryAfter, err := accounts.Get(ctx, credential.ID) + if err != nil { + t.Fatal(err) + } + if softRetryAfter.FailureCount != hard.FailureCount { + t.Fatalf("Retry-After soft failure count = %d, want preserved %d", softRetryAfter.FailureCount, hard.FailureCount) + } + if softRetryAfter.CooldownUntil == nil { + t.Fatal("Retry-After soft failure did not set cooldown") + } + cooldown = softRetryAfter.CooldownUntil.Sub(before) + if cooldown < 11*time.Second || cooldown > 13*time.Second { + t.Fatalf("Retry-After soft cooldown = %s, want ~12s", cooldown) + } + + before = time.Now().UTC() + selector.MarkFailure(ctx, softRetryAfter, 0, 0) preserved, err := accounts.Get(ctx, credential.ID) if err != nil { t.Fatal(err) diff --git a/backend/internal/application/gateway/service.go b/backend/internal/application/gateway/service.go index 3fae96164..702a62160 100644 --- a/backend/internal/application/gateway/service.go +++ b/backend/internal/application/gateway/service.go @@ -1546,8 +1546,13 @@ attemptLoop: if lastFailure.AccountScoped && !failureHandled { s.selector.MarkFailure(ctx, credential, response.StatusCode, retryAfter) } else if !lastFailure.AccountScoped && response.StatusCode >= http.StatusInternalServerError { - // 5xx 短冷却：本请求已 excluded，跨请求避免立刻再打同一坏号。 - s.selector.MarkFailure(ctx, credential, response.StatusCode, retryAfter) + // Provider-wide 5xx responses should rotate this request and briefly + // isolate the account across requests, but must not grow the durable + // account failure count exponentially. Preserve the real status in + // health diagnostics while applying the explicit soft policy. + if markErr := s.selector.markSoftFailure(ctx, credential, response.StatusCode, retryAfter); markErr != nil { + s.logger.Warn("upstream_soft_cooldown_failed", "request_id", input.RequestID, "account_id", credential.ID, "provider", credential.Provider, "status", response.StatusCode, "error", markErr) + } } lease.Release() lastErr = fmt.Errorf("上游返回 %d", response.StatusCode) diff --git a/backend/internal/application/gateway/service_test.go b/backend/internal/application/gateway/service_test.go index 95350dad5..173633273 100644 --- a/backend/internal/application/gateway/service_test.go +++ b/backend/internal/application/gateway/service_test.go @@ -3299,6 +3299,88 @@ func TestGatewayGeneric429CoolsAccountAndRotates(t *testing.T) { } } +func TestGatewayNonAccount5xxSoftCoolsAndRotates(t *testing.T) { + ctx := context.Background() + database, err := relational.OpenSQLite(ctx, filepath.Join(t.TempDir(), "soft-5xx.db")) + if err != nil { + t.Fatal(err) + } + defer database.Close() + if err := database.InitializeSchema(ctx); err != nil { + t.Fatal(err) + } + accountRepo := relational.NewAccountRepository(database) + modelRepo := relational.NewModelRepository(database) + auditRepo := relational.NewAuditRepository(database) + responseRepo := relational.NewResponseRepository(database) + keyRepo := relational.NewClientKeyRepository(database) + + credentials := make([]account.Credential, 0, 2) + for index, name := range []string{"gateway-a", "gateway-b"} { + credential, _, createErr := accountRepo.UpsertByIdentity(ctx, account.Credential{ + Provider: account.ProviderBuild, Name: name, SourceKey: name, EncryptedAccessToken: name, + ExpiresAt: time.Now().Add(time.Hour), Enabled: true, AuthStatus: account.AuthStatusActive, + Priority: 200 - index, MaxConcurrent: 1, + }) + if createErr != nil { + t.Fatal(createErr) + } + credentials = append(credentials, credential) + } + if err := accountRepo.UpdateHealth(ctx, credentials[0].ID, account.ProviderBuild, 3, nil, "prior account failure", false); err != nil { + t.Fatal(err) + } + if err := modelRepo.UpsertDiscovered(ctx, account.ProviderBuild, []string{"grok-soft-5xx"}); err != nil { + t.Fatal(err) + } + for _, credential := range credentials { + if err := modelRepo.ReplaceAccountCapabilities(ctx, credential.ID, []string{"grok-soft-5xx"}, time.Now().UTC()); err != nil { + t.Fatal(err) + } + } + clientKey, err := keyRepo.Create(ctx, clientkey.Key{ + Name: "soft-5xx-key", Prefix: "soft5xx", SecretHash: strings.Repeat("3", 64), EncryptedSecret: "encrypted", + Enabled: true, RPMLimit: 120, MaxConcurrent: 8, + }) + if err != nil { + t.Fatal(err) + } + adapter := &scriptedBuildAdapter{responses: map[uint64][]scriptedBuildResponse{ + credentials[0].ID: {{status: http.StatusGatewayTimeout, body: `{"error":"temporary upstream timeout"}`}}, + credentials[1].ID: {{status: http.StatusOK, body: `{"id":"resp-soft-5xx-b"}`}}, + }} + registry := provider.NewRegistry(adapter) + sticky := memory.NewStickyStore() + accountService := accountapp.NewService(accountRepo, auditRepo, memory.NewDeviceSessionStore(), sticky, registry, testCipher(t), nil) + selector := NewSelector(accountRepo, memory.NewConcurrencyLimiter(), sticky, registry, time.Hour, 30*time.Second, 30*time.Minute) + service := NewService(modelRepo, auditRepo, accountService, clientkeyapp.NewService(nil, nil, nil, 60, 4, nil), registry, selector, responseRepo, 3) + + before := time.Now().UTC() + result, err := service.CreateResponse(ctx, Input{ + RequestID: "req-soft-5xx", ClientKey: clientKey, PublicModel: "grok-soft-5xx", + Body: []byte(`{"model":"grok-soft-5xx","input":"hello"}`), + }) + if err != nil { + t.Fatal(err) + } + result.Finalize(Usage{}, "resp-soft-5xx-b", "") + _ = result.Body.Close() + if attempts := adapter.Attempts(); len(attempts) != 2 || attempts[0] != credentials[0].ID || attempts[1] != credentials[1].ID { + t.Fatalf("non-account 5xx must rotate, attempts=%#v", attempts) + } + cooled, err := accountRepo.Get(ctx, credentials[0].ID) + if err != nil { + t.Fatal(err) + } + if cooled.AuthStatus != account.AuthStatusActive || cooled.FailureCount != 3 || cooled.LastError != "upstream status 504" || cooled.CooldownUntil == nil { + t.Fatalf("non-account 5xx soft cooldown state = %#v", cooled) + } + cooldown := cooled.CooldownUntil.Sub(before) + if cooldown < 4*time.Second || cooldown > 6*time.Second { + t.Fatalf("non-account 5xx cooldown = %s, want ~5s", cooldown) + } +} + func TestGatewayExhausted429PreservesLastBodyInFailure(t *testing.T) { // When all attempts fail, CreateResponse returns UpstreamFailure (sanitized). // captureResponse must reattach the diagnostic body so subsequent classification diff --git a/backend/internal/infra/buildtransport/http2.go b/backend/internal/infra/buildtransport/http2.go new file mode 100644 index 000000000..d14901597 --- /dev/null +++ b/backend/internal/infra/buildtransport/http2.go @@ -0,0 +1,35 @@ +package buildtransport + +import ( + "errors" + "net/http" + "time" + + "golang.org/x/net/http2" +) + +const ( + // IdleConnTimeout stays below the CLI proxy's observed idle-close window so + // an idle connection is retired before a later POST can reuse it. + IdleConnTimeout = 30 * time.Second + // HTTP2ReadIdleTimeout periodically probes an otherwise idle HTTP/2 + // connection. Go's default is zero, which leaves half-dead pooled + // connections undetected until a request lands on them. + HTTP2ReadIdleTimeout = 20 * time.Second + HTTP2PingTimeout = 10 * time.Second +) + +// ConfigureHTTP2Health enables active PING health checks on a Build transport. +// It must be called after proxy and dialer options have been applied. +func ConfigureHTTP2Health(transport *http.Transport) (*http2.Transport, error) { + if transport == nil { + return nil, errors.New("Build HTTP transport is nil") + } + h2, err := http2.ConfigureTransports(transport) + if err != nil { + return nil, err + } + h2.ReadIdleTimeout = HTTP2ReadIdleTimeout + h2.PingTimeout = HTTP2PingTimeout + return h2, nil +} diff --git a/backend/internal/infra/buildtransport/http2_test.go b/backend/internal/infra/buildtransport/http2_test.go new file mode 100644 index 000000000..50e2fa4f9 --- /dev/null +++ b/backend/internal/infra/buildtransport/http2_test.go @@ -0,0 +1,26 @@ +package buildtransport + +import ( + "net/http" + "testing" +) + +func TestConfigureHTTP2HealthEnablesActivePing(t *testing.T) { + transport := &http.Transport{ForceAttemptHTTP2: true} + h2, err := ConfigureHTTP2Health(transport) + if err != nil { + t.Fatal(err) + } + if h2.ReadIdleTimeout != HTTP2ReadIdleTimeout || h2.PingTimeout != HTTP2PingTimeout { + t.Fatalf("HTTP/2 health = (%s, %s)", h2.ReadIdleTimeout, h2.PingTimeout) + } + if transport.TLSNextProto["h2"] == nil { + t.Fatal("HTTP/2 transport was not installed") + } +} + +func TestConfigureHTTP2HealthRejectsNilTransport(t *testing.T) { + if _, err := ConfigureHTTP2Health(nil); err == nil { + t.Fatal("nil transport was accepted") + } +} diff --git a/backend/internal/infra/egress/buildclient.go b/backend/internal/infra/egress/buildclient.go index d01a5b1db..442296f86 100644 --- a/backend/internal/infra/egress/buildclient.go +++ b/backend/internal/infra/egress/buildclient.go @@ -10,6 +10,7 @@ import ( "time" _ "github.com/bdandy/go-socks4" + "github.com/chenyme/grok2api/backend/internal/infra/buildtransport" "github.com/chenyme/grok2api/backend/internal/pkg/tunnelproxy" xproxy "golang.org/x/net/proxy" ) @@ -37,7 +38,7 @@ func newBuildClientWithOptions(proxyURL string, responseHeaderTimeout time.Durat MaxIdleConns: 256, MaxIdleConnsPerHost: 128, MaxConnsPerHost: 256, - IdleConnTimeout: 90 * time.Second, + IdleConnTimeout: buildtransport.IdleConnTimeout, TLSHandshakeTimeout: 10 * time.Second, ResponseHeaderTimeout: responseHeaderTimeout, ExpectContinueTimeout: time.Second, @@ -69,6 +70,9 @@ func newBuildClientWithOptions(proxyURL string, responseHeaderTimeout time.Durat return nil, fmt.Errorf("Grok Build 不支持代理协议 %q", parsed.Scheme) } } + if _, err := buildtransport.ConfigureHTTP2Health(transport); err != nil { + return nil, fmt.Errorf("配置 Grok Build HTTP/2 健康探测: %w", err) + } return &http.Client{ Transport: transport, CheckRedirect: func(*http.Request, []*http.Request) error { diff --git a/backend/internal/infra/egress/buildclient_test.go b/backend/internal/infra/egress/buildclient_test.go index a0ceaec60..9385b3544 100644 --- a/backend/internal/infra/egress/buildclient_test.go +++ b/backend/internal/infra/egress/buildclient_test.go @@ -13,6 +13,7 @@ import ( "time" application "github.com/chenyme/grok2api/backend/internal/application/egress" + "github.com/chenyme/grok2api/backend/internal/infra/buildtransport" neterrorpkg "github.com/chenyme/grok2api/backend/internal/pkg/neterror" ) @@ -25,6 +26,9 @@ func TestBuildClientUsesConfiguredResponseHeaderTimeout(t *testing.T) { if transport.ResponseHeaderTimeout != 7*time.Minute { t.Fatalf("response header timeout = %s", transport.ResponseHeaderTimeout) } + if transport.IdleConnTimeout != buildtransport.IdleConnTimeout || transport.TLSNextProto["h2"] == nil { + t.Fatalf("Build HTTP/2 health transport not configured: %#v", transport) + } } func TestBuildEnvironmentClientPreservesEnvironmentProxyLookup(t *testing.T) { @@ -109,7 +113,7 @@ func TestNewBuildClientUsesStandardTransportForEveryProxyFamily(t *testing.T) { if !ok { t.Fatalf("transport = %T, want *http.Transport", client.Transport) } - if transport.ForceAttemptHTTP2 != true || transport.DialContext == nil { + if transport.ForceAttemptHTTP2 != true || transport.DialContext == nil || transport.TLSNextProto["h2"] == nil { t.Fatalf("standard transport not fully configured: %#v", transport) } if (transport.Proxy != nil) != test.httpProxy { diff --git a/backend/internal/infra/provider/cli/adapter.go b/backend/internal/infra/provider/cli/adapter.go index 49b5afef6..e28dae943 100644 --- a/backend/internal/infra/provider/cli/adapter.go +++ b/backend/internal/infra/provider/cli/adapter.go @@ -24,6 +24,7 @@ import ( "github.com/chenyme/grok2api/backend/internal/domain/account" modeldomain "github.com/chenyme/grok2api/backend/internal/domain/model" settingsdomain "github.com/chenyme/grok2api/backend/internal/domain/settings" + "github.com/chenyme/grok2api/backend/internal/infra/buildtransport" infraegress "github.com/chenyme/grok2api/backend/internal/infra/egress" "github.com/chenyme/grok2api/backend/internal/infra/provider" "github.com/chenyme/grok2api/backend/internal/infra/provider/conversation" @@ -191,13 +192,17 @@ func (t *buildDirectTransport) UpdateResponseHeaderTimeout(responseHeaderTimeout } func newBuildHTTPTransport(responseHeaderTimeout time.Duration) *http.Transport { - return &http.Transport{ + transport := &http.Transport{ Proxy: http.ProxyFromEnvironment, ForceAttemptHTTP2: true, MaxIdleConns: 256, MaxIdleConnsPerHost: 128, MaxConnsPerHost: 256, - IdleConnTimeout: 90 * time.Second, TLSHandshakeTimeout: 10 * time.Second, + IdleConnTimeout: buildtransport.IdleConnTimeout, TLSHandshakeTimeout: 10 * time.Second, ResponseHeaderTimeout: normalizeBuildResponseHeaderTimeout(responseHeaderTimeout), ExpectContinueTimeout: time.Second, } + if _, err := buildtransport.ConfigureHTTP2Health(transport); err != nil { + slog.Warn("build_http2_health_config_failed", "error", err) + } + return transport } func normalizeBuildResponseHeaderTimeout(value time.Duration) time.Duration { diff --git a/backend/internal/infra/provider/cli/adapter_test.go b/backend/internal/infra/provider/cli/adapter_test.go index ab26634ae..4c8573d82 100644 --- a/backend/internal/infra/provider/cli/adapter_test.go +++ b/backend/internal/infra/provider/cli/adapter_test.go @@ -19,6 +19,7 @@ import ( "github.com/chenyme/grok2api/backend/internal/domain/account" modeldomain "github.com/chenyme/grok2api/backend/internal/domain/model" settingsdomain "github.com/chenyme/grok2api/backend/internal/domain/settings" + "github.com/chenyme/grok2api/backend/internal/infra/buildtransport" infraegress "github.com/chenyme/grok2api/backend/internal/infra/egress" "github.com/chenyme/grok2api/backend/internal/infra/provider" "github.com/chenyme/grok2api/backend/internal/infra/provider/conversation" @@ -33,6 +34,16 @@ func (fn roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) return fn(request) } +func TestBuildDirectTransportEnablesHTTP2Health(t *testing.T) { + transport := newBuildHTTPTransport(5 * time.Minute) + if transport.IdleConnTimeout != buildtransport.IdleConnTimeout { + t.Fatalf("idle connection timeout = %s", transport.IdleConnTimeout) + } + if transport.TLSNextProto["h2"] == nil { + t.Fatal("Build direct transport did not install HTTP/2 health checks") + } +} + func TestResponseRequestForcedEgressOverridesCredentialBinding(t *testing.T) { var gotNodeID uint64 adapter := NewAdapter(Config{BaseURL: "https://cli-chat-proxy.grok.com/v1"}, nil)