fix(health): 放宽上游错误重试并保留健康边界 (#643)

* fix(health): 放宽上游错误重试并保留健康边界

* fix(health): 补齐资源操作重放与上下文拒绝规则

* fix(execution): 保留明确请求拒绝的分类优先级

* fix(health): 补齐流内 HTTP 错误重试判断
This commit is contained in:
tbphp authored and GitHub committed 2026-09-13 18:06:37 +08:00
1 parent 5f11d8bdd4
commit fbb99067cc
18 files changed
+489 -62

No files matched your search

@@ -19,8 +19,14 @@ func TestCandidateCapabilityRejectionsAreNarrow(t *testing.T) {
{name: "unsupported model legacy spelling", status: 400, body: `{"error":{"type":"invalid_request_error","code":"unsupported-model"}}`, want: execution.FailureHintModelUnavailable},
{name: "known function calling rejection", status: 400, body: `{"error":{"type":"invalid_request_error","message":"function calling is not supported with this model"}}`, want: execution.FailureHintModelUnavailable},
{name: "context limit", status: 400, body: `{"error":{"type":"invalid_request_error","code":"context_length_exceeded"}}`, want: execution.FailureHintRequestRejected},
{name: "generic invalid request", status: 400, body: `{"error":{"type":"invalid_request_error","message":"invalid request"}}`, want: execution.FailureHintRequestRejected},
{name: "quoted message is not rejection evidence", status: 400, body: `{"error":{"type":"invalid_request_error","message":"invalid input: function calling is not supported with this model"}}`, want: execution.FailureHintRequestRejected},
{name: "Anthropic context limit", status: 400, body: `{"error":{"type":"invalid_request_error","message":"prompt is too long"}}`, want: execution.FailureHintRequestRejected},
{name: "Anthropic context limit with counts", status: 400, body: `{"error":{"type":"invalid_request_error","message":"prompt is too long: 210000 tokens > 200000 maximum"}}`, want: execution.FailureHintRequestRejected},
{name: "quoted context limit", status: 400, body: `{"error":{"type":"invalid_request_error","message":"provider diagnostic: prompt is too long"}}`},
{name: "context near match", status: 400, body: `{"error":{"type":"invalid_request_error","message":"prompt is too long-lived"}}`},
{name: "generic invalid request", status: 400, body: `{"error":{"type":"invalid_request_error","message":"invalid request"}}`},
{name: "quoted message is not rejection evidence", status: 400, body: `{"error":{"type":"invalid_request_error","message":"invalid input: function calling is not supported with this model"}}`},
{name: "quoted error code is not rejection evidence", status: 400, body: `{"error":{"message":"upstream returned invalid_parameter"}}`},
{name: "content policy rejection", status: 400, body: `{"error":{"code":"content_policy_violation"}}`, want: execution.FailureHintRequestRejected},
{name: "explicit invalid parameter wins", status: 400, body: `{"error":{"type":"invalid_request_error","code":"invalid_parameter","message":"function calling is not supported with this model"}}`, want: execution.FailureHintRequestRejected},
{name: "generic forbidden", status: http.StatusForbidden, body: `{"error":{"code":"permission_denied"}}`},
{name: "resource missing", status: http.StatusNotFound, body: `{"error":{"code":"response_not_found"}}`},
+2 -9
View File
@@ -298,6 +298,8 @@ func neutralFailureHint(status int, values ...string) execution.FailureHint {
switch {
case status == http.StatusUnauthorized:
return execution.FailureHintInvalidCredential
case len(values) >= 2 && execution.ExplicitRequestRejection(values[0], values[1], values[len(values)-1]):
return execution.FailureHintRequestRejected
case candidateCapabilityRejected(status, values):
return execution.FailureHintModelUnavailable
case containsAnyMarker(markers,
@@ -313,9 +315,6 @@ func neutralFailureHint(status int, values ...string) execution.FailureHint {
"authentication failed",
"invalid credential", "api key not valid"):
return execution.FailureHintInvalidCredential
case containsAnyMarker(markers,
"invalid_request_error", "context_length_exceeded"):
return execution.FailureHintRequestRejected
case containsAnyMarker(markers,
"server_is_overloaded", "service_unavailable", "server overloaded"):
return execution.FailureHintHostError
@@ -387,12 +386,6 @@ func annotateBifrostErrorEvidence(evidence *execution.ErrorEvidence) {
evidence.ScopeHint = execution.ErrorScopeGroup
case execution.ErrorKindCanceled, execution.ErrorKindInvalidRequest:
evidence.ScopeHint = execution.ErrorScopeRequest
case execution.ErrorKindHTTP:
if evidence.StatusCode >= http.StatusBadRequest &&
evidence.StatusCode < http.StatusInternalServerError &&
evidence.StatusCode != http.StatusTooManyRequests {
evidence.ScopeHint = execution.ErrorScopeRequest
}
}
}
}
+2 -4
View File
@@ -67,13 +67,11 @@ func TestPassthroughHTTPErrorProducesNeutralFailureHints(t *testing.T) {
},
{
name: "generic forbidden permission", status: http.StatusForbidden,
body: `{"error":{"code":"permission_denied"}}`,
scope: execution.ErrorScopeRequest,
body: `{"error":{"code":"permission_denied"}}`,
},
{
name: "payment required", status: http.StatusPaymentRequired,
body: `{"error":{"message":"billing disabled"}}`,
scope: execution.ErrorScopeRequest,
body: `{"error":{"message":"billing disabled"}}`,
},
{
name: "explicit invalid key under forbidden", status: http.StatusForbidden,
@@ -62,6 +62,17 @@ type classifiedClaudeError struct {
summary string
}
func TestClaudeContextLimitIsAnExplicitRequestRejection(t *testing.T) {
bridge := newClaudeProviderBridge()
_, evidence := bridge.ClassifyError(t.Context(), &classifiedClaudeError{
status: http.StatusBadRequest, typeValue: "invalid_request_error",
summary: "prompt is too long: 210000 tokens > 200000 maximum",
}, claudeProviderCredential{})
if evidence == nil || evidence.Hint != execution.FailureHintRequestRejected || evidence.ScopeHint != execution.ErrorScopeRequest {
t.Fatalf("context limit evidence=%#v, want request rejection", evidence)
}
}
func (err *classifiedClaudeError) Error() string { return err.summary }
func (err *classifiedClaudeError) StatusCode() int { return err.status }
func (err *classifiedClaudeError) ErrorType() string { return err.typeValue }
@@ -78,7 +78,7 @@ func TestCodexUpgradePreservesQuotaAndRejectionPolicy(t *testing.T) {
events := 0
result := adapter.ExecuteStream(ctx, spec, func(execution.StreamEvent) error { events++; return nil })
evidence, status, dispatch = result.Error, result.StatusCode, result.DispatchState
if test.wantReplay && events != 0 {
if events != 0 {
t.Fatalf("rejection emitted %d downstream events", events)
}
} else {
@@ -97,8 +97,8 @@ func TestCodexUpgradePreservesQuotaAndRejectionPolicy(t *testing.T) {
if decision.Effect != test.wantEffect {
t.Fatalf("health decision=%+v, want effect %s", decision, test.wantEffect)
}
if got := decision.Retry == health.RetryNextCandidate; got != test.wantReplay {
t.Fatalf("health retry=%s, want retryable=%t", decision.Retry, test.wantReplay)
if decision.Retry != health.RetryNextCandidate {
t.Fatalf("health retry=%s, want next candidate without changing replay evidence", decision.Retry)
}
})
}
-2
View File
@@ -231,8 +231,6 @@ func (*grokProviderBridge) ClassifyError(
case status == http.StatusForbidden || status == http.StatusPaymentRequired:
evidence.Hint = execution.FailureHintCandidateUnavailable
evidence.ReplaySafety = execution.ReplaySafetyRejectedBeforeProcessing
case status == http.StatusBadRequest:
evidence.Hint = execution.FailureHintRequestRejected
case status == http.StatusTooManyRequests:
evidence.Hint = execution.FailureHintRateLimited
case status >= http.StatusInternalServerError:
+4 -1
View File
@@ -40,7 +40,10 @@ func TestGrokProviderClassifiesOAuthAndQuotaFailures(t *testing.T) {
{name: "unauthorized", err: grokProviderTestError{status: http.StatusUnauthorized}, hint: execution.FailureHintRefreshRequired, scope: execution.ErrorScopeCredential, replay: execution.ReplaySafetyRejectedBeforeProcessing},
{name: "forbidden", err: grokProviderTestError{status: http.StatusForbidden}, hint: execution.FailureHintCandidateUnavailable, scope: execution.ErrorScopeModel, replay: execution.ReplaySafetyRejectedBeforeProcessing},
{name: "payment required", err: grokProviderTestError{status: http.StatusPaymentRequired}, hint: execution.FailureHintCandidateUnavailable, scope: execution.ErrorScopeModel, replay: execution.ReplaySafetyRejectedBeforeProcessing},
{name: "invalid request", err: grokProviderTestError{status: http.StatusBadRequest}, hint: execution.FailureHintRequestRejected, scope: execution.ErrorScopeRequest},
{name: "generic bad request", err: grokProviderTestError{status: http.StatusBadRequest}},
{name: "explicit invalid parameter", err: grokProviderTestError{status: http.StatusBadRequest, code: "invalid_parameter"}, hint: execution.FailureHintRequestRejected, scope: execution.ErrorScopeRequest},
{name: "unsupported model", err: grokProviderTestError{status: http.StatusBadRequest, code: "unsupported_model"}, hint: execution.FailureHintModelUnavailable, scope: execution.ErrorScopeModel},
{name: "unsupported model alias", err: grokProviderTestError{status: http.StatusBadRequest, code: "unsupported-model"}, hint: execution.FailureHintModelUnavailable, scope: execution.ErrorScopeModel},
{name: "free usage", err: grokProviderTestError{status: http.StatusTooManyRequests, code: "subscription:free-usage-exhausted", retry: 24 * time.Hour}, hint: execution.FailureHintRateLimited, scope: execution.ErrorScopeCredential},
{name: "host", err: grokProviderTestError{status: http.StatusServiceUnavailable}, hint: execution.FailureHintHostError, scope: execution.ErrorScopeGroup},
} {
+7 -2
View File
@@ -53,9 +53,14 @@ func annotateProviderErrorEvidence(evidence *execution.ErrorEvidence, err error)
if evidence == nil {
return
}
explicitRequestRejection := execution.ExplicitRequestRejection(evidence.Type, evidence.Code, evidence.Summary)
if evidence.Hint == "" && explicitRequestRejection {
evidence.Hint = execution.FailureHintRequestRejected
}
if evidence.Kind == execution.ErrorKindHTTP && evidence.StatusCode == http.StatusBadRequest &&
evidence.Hint == execution.FailureHintRequestRejected && !requestScopedFailure(err) {
// 通用 HTTP 400 标签不能盖过明确的模型拒绝;保留 provider 的强请求级判定。
(evidence.Hint == "" || evidence.Hint == execution.FailureHintRequestRejected) &&
!explicitRequestRejection && !requestScopedFailure(err) {
// 仅细化通用拒绝;明确参数错误和 provider 的强请求级判定优先。
switch strings.ToLower(strings.TrimSpace(evidence.Code)) {
case "unsupported_model", "unsupported-model":
evidence.Hint = execution.FailureHintModelUnavailable
+19 -6
View File
@@ -157,22 +157,35 @@ var _ providerBridge = (*recordingProviderBridge)(nil)
func TestProviderModelRejectionRefinesGenericRequestHint(t *testing.T) {
for _, test := range []struct {
name string
err error
want execution.FailureHint
name string
err error
hint execution.FailureHint
typeValue string
code string
want execution.FailureHint
}{
{name: "generic provider rejection", err: errors.New("model unsupported"), want: execution.FailureHintModelUnavailable},
{name: "explicit request scope preserved", err: requestScopedTestError{}, want: execution.FailureHintRequestRejected},
{name: "generic provider rejection", err: errors.New("model unsupported"), hint: execution.FailureHintRequestRejected, code: "unsupported_model", want: execution.FailureHintModelUnavailable},
{name: "model without generic hint", code: "unsupported_model", want: execution.FailureHintModelUnavailable},
{name: "model alias without generic hint", code: "unsupported-model", want: execution.FailureHintModelUnavailable},
{name: "explicit parameter error precedes model code", typeValue: "invalid_parameter", code: "unsupported_model", want: execution.FailureHintRequestRejected},
{name: "explicit request scope preserved", err: requestScopedTestError{}, hint: execution.FailureHintRequestRejected, code: "unsupported_model", want: execution.FailureHintRequestRejected},
} {
t.Run(test.name, func(t *testing.T) {
evidence := &execution.ErrorEvidence{
Kind: execution.ErrorKindHTTP, StatusCode: http.StatusBadRequest,
Hint: execution.FailureHintRequestRejected, Code: "unsupported_model",
Hint: test.hint, Type: test.typeValue, Code: test.code,
}
annotateProviderErrorEvidence(evidence, test.err)
if evidence.Hint != test.want {
t.Fatalf("hint = %s, want %s", evidence.Hint, test.want)
}
wantScope := execution.ErrorScopeModel
if test.want == execution.FailureHintRequestRejected {
wantScope = execution.ErrorScopeRequest
}
if evidence.ScopeHint != wantScope {
t.Fatalf("scope = %s, want %s", evidence.ScopeHint, wantScope)
}
})
}
}
+19
View File
@@ -0,0 +1,19 @@
package execution
import "strings"
// ExplicitRequestRejection 识别具体代码及明确的上下文超限,不以通用错误类型或任意消息子串终止重试。
func ExplicitRequestRejection(typeValue, codeValue, message string) bool {
for _, value := range []string{typeValue, codeValue} {
switch strings.ToLower(strings.TrimSpace(value)) {
case "context_length_exceeded", "invalid_parameter", "content_policy_violation":
return true
}
}
if !strings.EqualFold(strings.TrimSpace(typeValue), "invalid_request_error") || strings.TrimSpace(codeValue) != "" {
return false
}
message = strings.ToLower(strings.TrimSpace(message))
return strings.TrimSuffix(message, ".") == "prompt is too long" ||
strings.HasPrefix(message, "prompt is too long:")
}
@@ -682,7 +682,7 @@ func TestGatewayAppliesEachGroupsParameterOverridesFromOriginal(t *testing.T) {
}
}
func TestHandlerHostFailureDoesNotReplayGenerationAcrossGroups(t *testing.T) {
func TestHandlerHostFailureRetriesGenerationAcrossGroups(t *testing.T) {
primary := fakeupstream.New(
fakeupstream.Step{Status: http.StatusInternalServerError, Fixture: "openai/500.json"},
fakeupstream.Step{Status: http.StatusInternalServerError, Fixture: "openai/500.json"},
@@ -709,7 +709,7 @@ func TestHandlerHostFailureDoesNotReplayGenerationAcrossGroups(t *testing.T) {
request.Header.Set("Authorization", "Bearer gl-client")
recorder := httptest.NewRecorder()
engine.ServeHTTP(recorder, request)
if recorder.Code != http.StatusInternalServerError || recorder.Header().Get(debugHeaderAttempts) != "1" {
if recorder.Code != http.StatusOK || recorder.Header().Get(debugHeaderAttempts) != "2" {
t.Fatalf("response = %d attempts=%s body=%s",
recorder.Code, recorder.Header().Get(debugHeaderAttempts), recorder.Body.String())
}
@@ -717,8 +717,8 @@ func TestHandlerHostFailureDoesNotReplayGenerationAcrossGroups(t *testing.T) {
if got := len(primary.Requests()); got != 2 {
t.Fatalf("primary requests = %d, want one per downstream request", got)
}
if got := len(backup.Requests()); got != 0 {
t.Fatalf("backup requests = %d, want none for ambiguous generation failures", got)
if got := len(backup.Requests()); got != 2 {
t.Fatalf("backup requests = %d, want one per downstream request", got)
}
}
+218
View File
@@ -0,0 +1,218 @@
package gateway
import (
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"gpt-load/internal/channel"
"gpt-load/internal/dialect"
"gpt-load/internal/protocol"
)
func TestAnthropicPrechargeFailureRetriesWithoutCredentialPenalty(t *testing.T) {
for _, errorType := range []string{"", "invalid_request_error"} {
t.Run("type="+errorType, func(t *testing.T) {
var mu sync.Mutex
var credentials []string
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
key := r.Header.Get("X-Api-Key")
mu.Lock()
credentials = append(credentials, key)
mu.Unlock()
w.Header().Set("Content-Type", "application/json")
if key == "sk-one" {
w.WriteHeader(http.StatusForbidden)
_, _ = fmt.Fprintf(w, `{"type":"error","error":{"type":%q,"message":"预扣费额度失败,余额 0.10,需要预扣 0.20"}}`, errorType)
return
}
_, _ = w.Write([]byte(`{"id":"msg_test","type":"message","role":"assistant","model":"model-a","content":[{"type":"text","text":"ok"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}`))
}))
defer upstream.Close()
engine, registry := newDialectGatewayEngine(t, protocol.Anthropic, "model-a", dialect.NewSet(dialect.NewAnthropic()),
dialectGatewayGroup{id: 1, name: "precharge", upstreamURL: upstream.URL, apiKeys: []string{"sk-one", "sk-two"}},
)
request := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(`{"model":"model-a","max_tokens":1,"messages":[{"role":"user","content":"hello"}]}`))
request.Header.Set("Authorization", "Bearer gl-client")
response := httptest.NewRecorder()
engine.ServeHTTP(response, request)
mu.Lock()
defer mu.Unlock()
if response.Code != http.StatusOK || fmt.Sprint(credentials) != "[sk-one sk-two]" {
t.Fatalf("response=%d attempts=%v body=%s", response.Code, credentials, response.Body.String())
}
if candidates := registry.CollectCredentialCandidates([]uint{1}, nil, time.Now()); len(candidates) != 2 {
t.Fatalf("available credentials=%d, want both", len(candidates))
}
entries, err := registry.SnapshotGroupCredentialEntriesExact(1, []uint{1, 2})
if err != nil {
t.Fatal(err)
}
for _, entry := range entries {
if entry.FailureCount != 0 || entry.Blacklisted || !entry.CooldownUntil.IsZero() || len(entry.ModelCooldowns) != 0 {
t.Fatalf("credential %d unexpectedly penalized", entry.ID)
}
}
})
}
}
func TestUnknownResponsesCancellationFailureDoesNotSwitchCredential(t *testing.T) {
var attempts atomic.Int32
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
if attempts.Add(1) == 1 {
w.WriteHeader(http.StatusForbidden)
_, _ = fmt.Fprint(w, `{"error":{"message":"resource rejected"}}`)
return
}
_, _ = fmt.Fprint(w, `{"id":"resp_1","status":"cancelled"}`)
}))
defer upstream.Close()
engine, _ := newDialectGatewayEngine(t, protocol.OpenAIResponses, "model-a", dialect.NewSet(dialect.NewOpenAIResponses()),
dialectGatewayGroup{id: 1, name: "resource-retry", channelID: channel.OpenAI,
params: json.RawMessage(fmt.Sprintf(`{"base_url":%q}`, upstream.URL+"/v1")), apiKeys: []string{"sk-one", "sk-two"}},
)
request := httptest.NewRequest(http.MethodPost, "/v1/responses/resp_1/cancel", strings.NewReader("{}"))
request.Header.Set("Authorization", "Bearer gl-client")
response := httptest.NewRecorder()
engine.ServeHTTP(response, request)
if response.Code != http.StatusForbidden || attempts.Load() != 1 || !strings.Contains(response.Body.String(), "resource rejected") {
t.Fatalf("status=%d attempts=%d body=%s", response.Code, attempts.Load(), response.Body.String())
}
}
func TestUnknownErrorRetryRecordsAttemptsAndHonorsBudget(t *testing.T) {
for _, retryCount := range []int{0, 1} {
t.Run(fmt.Sprintf("retries=%d", retryCount), func(t *testing.T) {
failure := completeScriptedUpstreamResult(UpstreamResult{
StatusCode: http.StatusForbidden, RequestWritten: true,
Body: []byte(`{"error":{"message":"unclassified upstream rejection"}}`),
})
forwarder := &scriptedForwarder{results: []UpstreamResult{failure, successfulAffinityResult()}}
handler, manager, _ := newHandlerForTest(t, forwarder, "sk-one", "sk-two")
manager.Current().Settings.RetryCount = retryCount
sink := &recordingRequestLogSink{}
handler.requestLogSink = sink
engine := newAffinityTestEngine(t, handler)
request := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"gpt-4o","messages":[]}`))
request.Header.Set("Authorization", "Bearer gl-client")
response := httptest.NewRecorder()
engine.ServeHTTP(response, request)
events := sink.snapshot()
if len(events) != 1 || len(events[0].Attempts) != retryCount+1 || len(forwarder.inputs) != retryCount+1 {
t.Fatalf("attempts=%d events=%#v", len(forwarder.inputs), events)
}
first := events[0].Attempts[0]
if first.RetryDirective != "next_candidate" || first.Effect != "none" ||
first.WillRetry != (retryCount > 0) || first.FailureCategory != "ambiguous" {
t.Fatalf("first attempt=%#v", first)
}
if retryCount == 0 && (response.Code != http.StatusForbidden || response.Body.String() != string(failure.Body)) {
t.Fatalf("terminal response=%d %s", response.Code, response.Body.String())
}
if retryCount > 0 && (response.Code != http.StatusOK || forwarder.inputs[0].APIKey == forwarder.inputs[1].APIKey) {
t.Fatalf("retry did not switch credential: status=%d", response.Code)
}
})
}
}
func TestUnknownFirstStreamErrorRetriesBeforeOutput(t *testing.T) {
var mu sync.Mutex
var attempts []string
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
key := r.Header.Get("Authorization")
mu.Lock()
attempts = append(attempts, key)
mu.Unlock()
w.Header().Set("Content-Type", "text/event-stream")
if key == "Bearer sk-one" {
_, _ = fmt.Fprint(w, "data: "+`{"error":{"type":"new_provider_error","code":"unrecognized_condition","message":"first candidate failed"}}`+"\n\n")
return
}
_, _ = fmt.Fprint(w, "data: "+`{"id":"test","object":"chat.completion.chunk","model":"model-a","choices":[{"index":0,"delta":{"content":"recovered"},"finish_reason":"stop"}]}`+"\n\ndata: [DONE]\n\n")
}))
defer upstream.Close()
engine, _ := newDialectGatewayEngine(t, protocol.OpenAICompletions, "model-a", dialect.NewSet(dialect.NewOpenAI()),
dialectGatewayGroup{id: 1, name: "stream-retry", upstreamURL: upstream.URL, apiKeys: []string{"sk-one", "sk-two"}},
)
request := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"model-a","stream":true,"messages":[]}`))
request.Header.Set("Authorization", "Bearer gl-client")
response := httptest.NewRecorder()
engine.ServeHTTP(response, request)
mu.Lock()
defer mu.Unlock()
if response.Code != http.StatusOK || fmt.Sprint(attempts) != "[Bearer sk-one Bearer sk-two]" ||
!strings.Contains(response.Body.String(), "recovered") || strings.Contains(response.Body.String(), "unrecognized_condition") {
t.Fatalf("status=%d attempts=%v body=%s", response.Code, attempts, response.Body.String())
}
}
func TestConvertedStreamHTTPErrorRetriesBeforeOutput(t *testing.T) {
for _, test := range []struct {
name string
status int
partial bool
wantAttempts string
}{
{name: "server error skips failed group", status: 503, wantAttempts: "sk-one,sk-next"},
{name: "unknown rejection switches credential", status: 403, wantAttempts: "sk-one,sk-two"},
{name: "server error after output stops", status: 503, partial: true, wantAttempts: "sk-one"},
} {
t.Run(test.name, func(t *testing.T) {
var mu sync.Mutex
var attempts []string
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/v1beta/models/model-a:streamGenerateContent" {
t.Errorf("upstream path = %s", r.URL.Path)
}
key := r.Header.Get("X-Goog-Api-Key")
mu.Lock()
attempts = append(attempts, key)
mu.Unlock()
w.Header().Set("Content-Type", "text/event-stream")
if key == "sk-one" {
if test.partial {
_, _ = fmt.Fprint(w, "data: "+`{"candidates":[{"content":{"parts":[{"text":"partial"}],"role":"model"},"index":0}],"modelVersion":"model-a","responseId":"resp_1"}`+"\n\n")
}
_, _ = fmt.Fprintf(w, "data: {\"error\":{\"code\":%d,\"message\":\"first candidate failed\"}}\n\n", test.status)
return
}
_, _ = fmt.Fprint(w, "data: "+`{"candidates":[{"content":{"parts":[{"text":"recovered"}],"role":"model"},"finishReason":"STOP","index":0}],"modelVersion":"model-a","responseId":"resp_2","usageMetadata":{"promptTokenCount":1,"candidatesTokenCount":1,"totalTokenCount":2}}`+"\n\n")
}))
defer upstream.Close()
params, err := json.Marshal(map[string]string{"base_url": upstream.URL + "/v1beta"})
if err != nil {
t.Fatal(err)
}
engine, _ := newDialectGatewayEngine(t, protocol.Anthropic, "model-a", dialect.NewSet(dialect.NewAnthropic()),
dialectGatewayGroup{id: 1, name: "first", channelID: channel.Gemini, params: params, apiKeys: []string{"sk-one", "sk-two"}},
dialectGatewayGroup{id: 2, name: "backup", channelID: channel.Gemini, params: params, apiKeys: []string{"sk-next"}},
)
request := httptest.NewRequest(http.MethodPost, "/v1/messages", strings.NewReader(`{"model":"model-a","max_tokens":16,"stream":true,"messages":[{"role":"user","content":"hello"}]}`))
request.Header.Set("Authorization", "Bearer gl-client")
response := httptest.NewRecorder()
engine.ServeHTTP(response, request)
mu.Lock()
defer mu.Unlock()
if strings.Join(attempts, ",") != test.wantAttempts {
t.Fatalf("attempts=%v, want %s; status=%d body=%s", attempts, test.wantAttempts, response.Code, response.Body.String())
}
body := response.Body.String()
if test.partial {
if !strings.Contains(body, "partial") || strings.Contains(body, "recovered") {
t.Fatalf("committed stream response=%s", body)
}
} else if response.Code != http.StatusOK || !strings.Contains(body, "recovered") || strings.Contains(body, "first candidate failed") {
t.Fatalf("retry response=%d %s", response.Code, body)
}
})
}
}
+2 -2
View File
@@ -177,7 +177,7 @@ func TestJudgeUpstreamResultUsesStableStreamTerminalRules(t *testing.T) {
wantOrigin execution.ErrorOrigin
wantScope execution.ErrorScope
}{
{name: "provider error", endReason: StreamEndSSEError, wantRule: "stream.provider_error", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest},
{name: "provider error", endReason: StreamEndSSEError, wantRule: "safety.committed", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest},
{name: "upstream terminated", endReason: StreamEndUpstreamTerminated, wantRule: "stream.upstream_terminated", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest},
{name: "protocol error", endReason: StreamEndUpstreamProtocolError, wantRule: "stream.protocol_error", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest},
{name: "idle timeout", endReason: StreamEndIdleTimeout, wantRule: "stream.idle_timeout", wantOrigin: execution.ErrorOriginUpstream, wantScope: execution.ErrorScopeRequest},
@@ -192,7 +192,7 @@ func TestJudgeUpstreamResultUsesStableStreamTerminalRules(t *testing.T) {
StatusCode: http.StatusOK,
Committed: true,
Stream: streamTerminalObservation(test.endReason),
}, time.Now(), health.DecisionContext{})
}, time.Now(), health.DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion})
if decision.Category != health.FailureCategoryAmbiguous ||
decision.Origin != test.wantOrigin || decision.Scope != test.wantScope ||
decision.Retry != health.RetryNone || decision.Effect != health.EffectNone ||
+5 -4
View File
@@ -3303,8 +3303,9 @@ func TestHandlerUsesClassifierForNonStreamingNonSuccess(t *testing.T) {
t.Run("client error terminates after one attempt", func(t *testing.T) {
forwarder := &scriptedForwarder{results: []UpstreamResult{
{StatusCode: http.StatusBadRequest, Header: make(http.Header),
Body: []byte(`{"error":"invalid input"}`),
ClassificationBody: []byte(`{"error":"invalid input"}`), RequestWritten: true},
Body: []byte(`{"error":"invalid input"}`),
ExecutionError: &execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Code: "invalid_parameter"},
RequestWritten: true},
{StatusCode: http.StatusOK, Header: make(http.Header), Body: []byte(`{"ok":true}`)},
}}
engine, _, _ := newHandlerTestRuntime(t, forwarder, "sk-one", "sk-two")
@@ -4647,8 +4648,8 @@ func TestHandlerDoesNotExposeAliasedUpstreamModelWhenRetryBudgetIsExhausted(t *t
engine.ServeHTTP(recorder, request)
if recorder.Code != http.StatusInternalServerError || attempts.Load() != 1 {
t.Fatalf("response/attempts = %d/%d, want 500/1", recorder.Code, attempts.Load())
if recorder.Code != http.StatusInternalServerError || attempts.Load() != 3 {
t.Fatalf("response/attempts = %d/%d, want 500/3", recorder.Code, attempts.Load())
}
if strings.Contains(recorder.Body.String(), upstreamModel) ||
!strings.Contains(recorder.Body.String(), externalModel) {
+47 -6
View File
@@ -297,6 +297,9 @@ func classifyExecutionEvidence(attempt ExecutionAttempt) FailureCategory {
case execution.FailureHintHostError:
return FailureCategoryUpstreamHostError
}
if execution.ExplicitRequestRejection(attempt.Evidence.Type, attempt.Evidence.Code, attempt.Evidence.Summary) {
return FailureCategoryClientError
}
markers = strings.ToLower(strings.Join([]string{
attempt.Evidence.Type,
attempt.Evidence.Code,
@@ -321,8 +324,6 @@ func classifyExecutionEvidence(attempt ExecutionAttempt) FailureCategory {
return FailureCategoryInvalidKey
case statusCode >= http.StatusInternalServerError && statusCode <= 599:
return FailureCategoryUpstreamHostError
case statusCode >= http.StatusBadRequest && statusCode <= 499:
return FailureCategoryClientError
case attempt.Evidence != nil && attempt.Evidence.Kind == execution.ErrorKindInvalidRequest:
return FailureCategoryClientError
default:
@@ -409,10 +410,10 @@ func decisionForExecutionCategory(
if attempt.Evidence.ReplaySafety == execution.ReplaySafetyRejectedBeforeProcessing {
retry = RetryNextCandidate
ruleID = "upstream.host_error.rejected_before_processing"
} else if attempt.Evidence.ReplaySafety != execution.ReplaySafetyUnknown &&
} else if retryableUpstreamResponse(attempt) &&
requestMayReplayAfterResponse(decisionContext) {
retry = RetryNextCandidate
ruleID = "upstream.host_error.read_only"
ruleID = "upstream.host_error.response_retry"
}
return decision(
category,
@@ -443,7 +444,44 @@ func decisionForExecutionCategory(
case FailureCategoryOK:
return decision(category, origin, scope, RetryNone, EffectNone, "success.upstream_response")
default:
return decision(category, origin, scope, RetryNone, EffectNone, ambiguousRuleID(attempt.Evidence))
ruleID := ambiguousRuleID(attempt.Evidence)
if retryableUpstreamResponse(attempt) {
if !requestMayReplayAfterResponse(decisionContext) &&
attempt.Evidence.ReplaySafety != execution.ReplaySafetyRejectedBeforeProcessing {
return decision(category, origin, scope, RetryNone, EffectNone, "safety.operation_replay_unsafe")
}
if ruleID == "fallback.ambiguous" {
ruleID = "fallback.upstream_response"
}
return decision(category, origin, scope, RetryNextCandidate, EffectNone, ruleID)
}
return decision(category, origin, scope, RetryNone, EffectNone, ruleID)
}
}
// retryableUpstreamResponse 区分已收到的错误响应与超时、断流等执行结果未知的失败。
func retryableUpstreamResponse(attempt ExecutionAttempt) bool {
evidence := attempt.Evidence
if evidence == nil || originForEvidence(evidence) != execution.ErrorOriginUpstream ||
evidence.ReplaySafety == execution.ReplaySafetyUnknown {
return false
}
switch evidence.Kind {
case execution.ErrorKindHTTP:
status := attempt.StatusCode
if status == 0 {
status = evidence.StatusCode
}
// 流建立后仍保留外层 2xx,流内 HTTP 错误由 Kind/Hint 承载。
return isSuccessStatus(status) || status >= http.StatusBadRequest && status <= 599
case execution.ErrorKindProvider:
switch evidence.Code {
case "upstream_protocol_error", "upstream_response_incomplete":
return false
}
return attempt.ResponseStarted()
default:
return false
}
}
@@ -695,7 +733,10 @@ func requestMayReplayAfterResponse(value DecisionContext) bool {
return true
}
switch value.Operation {
case execution.OperationResponsesRetrieve,
case execution.OperationChatCompletion,
execution.OperationResponsesCreate,
execution.OperationResponsesCompact,
execution.OperationResponsesRetrieve,
execution.OperationResponsesInputItems,
execution.OperationResponsesInputTokens,
execution.OperationCountTokens,
+6 -6
View File
@@ -216,22 +216,22 @@ func TestJudgeExecutionUsesNeutralEvidenceAndReplayBoundary(t *testing.T) {
want: Result{Category: FailureCategoryAmbiguous, Action: ActionTerminate},
},
{
name: "payment required is a client error",
name: "payment required retries without penalty",
attempt: ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent,
StatusCode: http.StatusPaymentRequired,
Evidence: evidence(execution.ErrorKindHTTP, http.StatusPaymentRequired, "billing disabled"),
},
want: Result{Category: FailureCategoryClientError, Action: ActionTerminate},
want: Result{Category: FailureCategoryAmbiguous, Action: ActionRetry},
},
{
name: "generic forbidden is a client error",
name: "generic forbidden retries without penalty",
attempt: ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent,
StatusCode: http.StatusForbidden,
Evidence: evidence(execution.ErrorKindHTTP, http.StatusForbidden, "permission denied"),
},
want: Result{Category: FailureCategoryClientError, Action: ActionTerminate},
want: Result{Category: FailureCategoryAmbiguous, Action: ActionRetry},
},
{
name: "forbidden model marker retries without penalty",
@@ -316,13 +316,13 @@ func TestJudgeExecutionUsesNeutralEvidenceAndReplayBoundary(t *testing.T) {
},
},
{
name: "generic not found terminates",
name: "generic not found retries without penalty",
attempt: ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent,
StatusCode: http.StatusNotFound,
Evidence: evidence(execution.ErrorKindHTTP, http.StatusNotFound, "endpoint not found"),
},
want: Result{Category: FailureCategoryClientError, Action: ActionTerminate},
want: Result{Category: FailureCategoryAmbiguous, Action: ActionRetry},
},
{
name: "provider rate limit under success status still cools credential",
+13 -11
View File
@@ -289,7 +289,7 @@ func TestJudgeExecutionPreservesReplayCompatibilityRules(t *testing.T) {
wantRule RuleID
}{
{
name: "generic payment required terminates",
name: "generic payment required retries",
attempt: ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent,
StatusCode: http.StatusPaymentRequired,
@@ -298,8 +298,9 @@ func TestJudgeExecutionPreservesReplayCompatibilityRules(t *testing.T) {
Summary: "billing disabled",
},
},
wantRetry: RetryNone,
wantRule: RuleID("fallback.http_client_error"),
context: DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion},
wantRetry: RetryNextCandidate,
wantRule: RuleID("fallback.upstream_response"),
},
{
name: "candidate payment required retries",
@@ -340,10 +341,10 @@ func TestJudgeExecutionPreservesReplayCompatibilityRules(t *testing.T) {
},
context: DecisionContext{Method: http.MethodGet, Operation: execution.OperationResponsesRetrieve},
wantRetry: RetryNextCandidate,
wantRule: RuleID("upstream.host_error.read_only"),
wantRule: RuleID("upstream.host_error.response_retry"),
},
{
name: "mutating host error terminates",
name: "generation host error retries",
attempt: ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent,
StatusCode: http.StatusServiceUnavailable,
@@ -353,8 +354,8 @@ func TestJudgeExecutionPreservesReplayCompatibilityRules(t *testing.T) {
},
},
context: DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion},
wantRetry: RetryNone,
wantRule: RuleID("upstream.host_error.replay_unsafe"),
wantRetry: RetryNextCandidate,
wantRule: RuleID("upstream.host_error.response_retry"),
},
{
name: "explicitly rejected mutating host error retries",
@@ -496,13 +497,13 @@ func TestJudgeExecutionReplayUnknownKeepsUnaffectedRuleID(t *testing.T) {
wantRule: "rate_limit.scoped",
},
{
name: "generic client error",
name: "generic forbidden with unknown replay safety",
status: http.StatusForbidden,
evidence: execution.ErrorEvidence{
Kind: execution.ErrorKindHTTP, StatusCode: http.StatusForbidden,
ReplaySafety: execution.ReplaySafetyUnknown, Summary: "permission denied",
},
wantRule: "fallback.http_client_error",
wantRule: "safety.replay_unknown",
},
}
@@ -738,8 +739,9 @@ func TestJudgeExecutionCompatibilityRuleMatrix(t *testing.T) {
Kind: execution.ErrorKindHTTP, StatusCode: http.StatusForbidden,
Summary: "permission denied",
},
category: FailureCategoryClientError, scope: execution.ErrorScopeRequest,
retry: RetryNone, effect: EffectNone, ruleID: "fallback.http_client_error",
context: DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion},
category: FailureCategoryAmbiguous,
retry: RetryNextCandidate, effect: EffectNone, ruleID: "fallback.upstream_response",
},
{
name: "candidate scoped 403", status: http.StatusForbidden,
+119
View File
@@ -0,0 +1,119 @@
package health
import (
"net/http"
"testing"
"gpt-load/internal/execution"
)
func TestUnknownHTTPRejectionsRetryWithoutCredentialEffects(t *testing.T) {
for _, status := range []int{400, 402, 403, 404, 409, 422} {
for _, code := range []string{"", "unknown_provider_error"} {
result := JudgeExecution(ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent, StatusCode: status,
Evidence: &execution.ErrorEvidence{
Kind: execution.ErrorKindHTTP, StatusCode: status,
OriginHint: execution.ErrorOriginUpstream,
Type: "invalid_request_error", Code: code, Summary: "request could not be served",
},
}, DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion})
if result.Category != FailureCategoryAmbiguous || result.Retry != RetryNextCandidate ||
result.Effect != EffectNone || !result.CooldownUntil.IsZero() {
t.Errorf("status=%d code=%q decision=%#v", status, code, result)
}
}
}
}
func TestExplicitRequestRejectionCodesStillTerminate(t *testing.T) {
for _, code := range []string{"context_length_exceeded", "invalid_parameter", "content_policy_violation"} {
result := JudgeExecution(ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusBadRequest,
Evidence: &execution.ErrorEvidence{
Kind: execution.ErrorKindHTTP, Code: code, OriginHint: execution.ErrorOriginUpstream,
},
}, DecisionContext{Method: http.MethodPost, Operation: execution.OperationChatCompletion})
if result.Category != FailureCategoryClientError || result.Retry != RetryNone || result.Effect != EffectNone {
t.Errorf("code=%s decision=%#v", code, result)
}
}
}
func TestUpstreamResponseRetryPreservesExecutionBoundaries(t *testing.T) {
for _, test := range []struct {
name string
status int
evidence execution.ErrorEvidence
operation execution.Operation
committed bool
retry RetryDirective
effect Effect
rule RuleID
}{
{name: "generation server error", status: 503, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP}, retry: RetryNextCandidate, effect: EffectSkipGroup},
{name: "server error inside successful stream", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError}, retry: RetryNextCandidate, effect: EffectSkipGroup},
{name: "unknown HTTP error inside successful stream", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP}, retry: RetryNextCandidate, effect: EffectNone},
{name: "stream HTTP error with unknown outcome", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError, ReplaySafety: execution.ReplaySafetyUnknown}, retry: RetryNone, effect: EffectSkipGroup},
{name: "committed stream HTTP error", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError}, committed: true, retry: RetryNone, effect: EffectNone},
{name: "stream HTTP error retains strict replay policy", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError}, operation: execution.OperationImagesGenerate, retry: RetryNone, effect: EffectSkipGroup},
{name: "stream HTTP error retains resource boundary", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, Hint: execution.FailureHintHostError}, operation: execution.OperationResponsesCancel, retry: RetryNone, effect: EffectSkipGroup},
{name: "unknown error event", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "upstream_sse_error"}, retry: RetryNextCandidate, effect: EffectNone},
{name: "unknown provider code", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "new_provider_error"}, retry: RetryNextCandidate, effect: EffectNone},
{name: "response status in evidence", evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, StatusCode: 403}, retry: RetryNextCandidate, effect: EffectNone},
{name: "transport after headers", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindTransport}, retry: RetryNone, effect: EffectNone},
{name: "timeout after headers", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindTimeout}, retry: RetryNone, effect: EffectNone},
{name: "protocol failure", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "upstream_protocol_error"}, retry: RetryNone, effect: EffectNone},
{name: "incomplete response", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "upstream_response_incomplete"}, retry: RetryNone, effect: EffectNone},
{name: "internal result with HTTP status", status: 400, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindInternal}, retry: RetryNone, effect: EffectNone},
{name: "unknown execution result", status: 503, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, ReplaySafety: execution.ReplaySafetyUnknown}, retry: RetryNone, effect: EffectSkipGroup},
{name: "committed event", status: 200, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindProvider, Code: "upstream_sse_error"}, committed: true, retry: RetryNone, effect: EffectNone},
{name: "image needs rejection proof", status: 503, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP}, operation: execution.OperationImagesGenerate, retry: RetryNone, effect: EffectSkipGroup, rule: "upstream.host_error.replay_unsafe"},
{name: "resource mutation remains restricted", status: 503, evidence: execution.ErrorEvidence{Kind: execution.ErrorKindHTTP}, operation: execution.OperationResponsesCancel, retry: RetryNone, effect: EffectSkipGroup},
} {
t.Run(test.name, func(t *testing.T) {
operation := test.operation
if operation == "" {
operation = execution.OperationChatCompletion
}
result := JudgeExecution(ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent, StatusCode: test.status,
Evidence: &test.evidence, DownstreamCommitted: test.committed,
}, DecisionContext{Method: http.MethodPost, Operation: operation})
if result.Retry != test.retry || result.Effect != test.effect {
t.Fatalf("decision=%#v, want retry=%s effect=%s", result, test.retry, test.effect)
}
if test.rule != "" && result.RuleID != test.rule {
t.Fatalf("rule=%s, want %s", result.RuleID, test.rule)
}
})
}
}
func TestUnknownErrorRetryRespectsOperationReplayPolicy(t *testing.T) {
for _, test := range []struct {
operation execution.Operation
method string
}{
{execution.OperationResponsesDelete, http.MethodDelete},
{execution.OperationResponsesCancel, http.MethodPost},
{execution.OperationResponsesPassthrough, http.MethodPost},
{execution.OperationResponsesPassthrough, http.MethodGet},
{execution.OperationChatCompletion, http.MethodPost},
} {
for _, proof := range []execution.ReplaySafety{"", execution.ReplaySafetyRejectedBeforeProcessing} {
result := JudgeExecution(ExecutionAttempt{
DispatchState: execution.DispatchMaybeSent, StatusCode: http.StatusForbidden,
Evidence: &execution.ErrorEvidence{Kind: execution.ErrorKindHTTP, ReplaySafety: proof},
}, DecisionContext{Method: test.method, Operation: test.operation})
want := RetryNone
if test.method == http.MethodGet || test.operation == execution.OperationChatCompletion || proof != "" {
want = RetryNextCandidate
}
if result.Retry != want || result.Effect != EffectNone {
t.Errorf("operation=%s method=%s proof=%q decision=%#v, want retry=%s without effects",
test.operation, test.method, proof, result, want)
}
}
}
}