mirror of
https://github.com/router-for-me/CLIProxyAPI.git
synced 2026-10-20 06:29:50 +08:00
feat(auth): add retry logic for pre-HTTP transport failures and related tests
- Introduce `isTransientTransportError` to identify transient transport errors such as TLS handshake, DNS, and connection resets. - Extend retry logic to handle these errors without enforcing credential cooldown. - Add new constants, such as `ErrorCodeTransientTransport`, to categorize and manage these errors. - Refactor `conductor_selection` to distinguish request-retry round errors. - Add comprehensive tests covering transport retries, cooldown behavior, and transient error handling. Closes: #5829
This commit is contained in:
@@ -6,10 +6,12 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
@@ -1470,6 +1472,10 @@ func resultErrorFromError(err error) *Error {
|
||||
if resultErr.Code == "" || resultErr.Code == connectionLifecycleErrorCode {
|
||||
resultErr.Code = connectionLifecycleErrorCode
|
||||
}
|
||||
case isTransientTransportError(err):
|
||||
if resultErr.Code == "" || resultErr.Code == transientTransportErrorCode {
|
||||
resultErr.Code = transientTransportErrorCode
|
||||
}
|
||||
}
|
||||
return resultErr
|
||||
}
|
||||
@@ -1481,7 +1487,7 @@ func shouldSkipCredentialCooldown(err *Error) bool {
|
||||
if err != nil && err.Code == ErrorCodeForceCooldown {
|
||||
return false
|
||||
}
|
||||
return isRequestScopedResultError(err) || isConnectionLifecycleResultError(err)
|
||||
return isRequestScopedResultError(err) || isConnectionLifecycleResultError(err) || isTransientTransportResultError(err)
|
||||
}
|
||||
|
||||
// isConnectionLifecycleError reports transport/session lifecycle failures that must
|
||||
@@ -1546,6 +1552,95 @@ func isConnectionLifecycleMessage(message string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// isTransientTransportError reports pre-HTTP dial/TLS/DNS/reset failures that
|
||||
// should retry under request-retry without cooling the selected credential.
|
||||
func isTransientTransportError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
// HTTP-status failures stay on the credential/status retry path.
|
||||
if statusCodeFromError(err) != 0 {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
|
||||
return true
|
||||
}
|
||||
var dnsErr *net.DNSError
|
||||
if errors.As(err, &dnsErr) && dnsErr != nil && (dnsErr.IsTimeout || dnsErr.IsTemporary || dnsErr.Timeout()) {
|
||||
return true
|
||||
}
|
||||
var netErr net.Error
|
||||
if errors.As(err, &netErr) && netErr != nil && netErr.Timeout() {
|
||||
return true
|
||||
}
|
||||
if isTransientSyscallError(err) {
|
||||
return true
|
||||
}
|
||||
var opErr *net.OpError
|
||||
if errors.As(err, &opErr) && opErr != nil {
|
||||
return true
|
||||
}
|
||||
return isTransientTransportMessage(err.Error())
|
||||
}
|
||||
|
||||
func isTransientTransportResultError(err *Error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if err.Code == transientTransportErrorCode {
|
||||
return true
|
||||
}
|
||||
if statusCodeFromResult(err) != 0 {
|
||||
return false
|
||||
}
|
||||
return isTransientTransportMessage(err.Message)
|
||||
}
|
||||
|
||||
func isTransientSyscallError(err error) bool {
|
||||
var errno syscall.Errno
|
||||
if !errors.As(err, &errno) {
|
||||
return false
|
||||
}
|
||||
switch errno {
|
||||
case syscall.ECONNREFUSED, syscall.ECONNRESET, syscall.ECONNABORTED,
|
||||
syscall.ETIMEDOUT, syscall.EHOSTUNREACH, syscall.ENETUNREACH, syscall.EPIPE:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func isTransientTransportMessage(message string) bool {
|
||||
lower := strings.ToLower(strings.TrimSpace(message))
|
||||
if lower == "" {
|
||||
return false
|
||||
}
|
||||
switch {
|
||||
case strings.Contains(lower, "tls: tls handshake"),
|
||||
strings.Contains(lower, "tls handshake timeout"),
|
||||
strings.Contains(lower, "wsarecv"),
|
||||
strings.Contains(lower, "wsasend"),
|
||||
strings.Contains(lower, "a connection attempt failed"),
|
||||
strings.Contains(lower, "connection refused"),
|
||||
strings.Contains(lower, "connection reset"),
|
||||
strings.Contains(lower, "i/o timeout"),
|
||||
strings.Contains(lower, "no such host"),
|
||||
strings.Contains(lower, "server misbehaving"),
|
||||
strings.Contains(lower, "network is unreachable"),
|
||||
strings.Contains(lower, "no route to host"),
|
||||
strings.Contains(lower, "broken pipe"),
|
||||
strings.Contains(lower, "connection aborted"),
|
||||
strings.Contains(lower, "use of closed network connection"),
|
||||
strings.Contains(lower, "unexpected eof"):
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func isUnauthorizedError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
|
||||
@@ -1302,7 +1302,7 @@ func (m *Manager) shouldRetryAfterErrorWithAttempted(ctx context.Context, opts c
|
||||
}
|
||||
var exhausted *homeRetryRoundExhaustedError
|
||||
if m.HomeEnabled() && errors.As(err, &exhausted) && exhausted != nil {
|
||||
if !isCredentialRetryRoundStatus(status) || !m.homeRetryAllowed(attempt, homeRetryLimit) {
|
||||
if !isRequestRetryRoundError(err) || !m.homeRetryAllowed(attempt, homeRetryLimit) {
|
||||
return 0, false
|
||||
}
|
||||
if exhausted.retryNow {
|
||||
@@ -1330,7 +1330,7 @@ func (m *Manager) shouldRetryAfterErrorWithAttempted(ctx context.Context, opts c
|
||||
}
|
||||
eligibility := authSelectionEligibilityForRequest(ctx, opts)
|
||||
pinnedAuthID := pinnedAuthIDFromMetadata(opts.Metadata)
|
||||
if !isCredentialRetryRoundStatus(status) || !m.retryAllowed(attempt, providers, model, eligibility, pinnedAuthID, defaultRequestRetry) {
|
||||
if !isRequestRetryRoundError(err) || !m.retryAllowed(attempt, providers, model, eligibility, pinnedAuthID, defaultRequestRetry) {
|
||||
return 0, false
|
||||
}
|
||||
wait, found := m.closestCooldownWaitWithAttempted(providers, model, attempt, eligibility, pinnedAuthID, defaultRequestRetry, status, attempted)
|
||||
@@ -1409,6 +1409,13 @@ func isCredentialRetryRoundStatus(status int) bool {
|
||||
}
|
||||
}
|
||||
|
||||
func isRequestRetryRoundError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
return isCredentialRetryRoundStatus(statusCodeFromError(err)) || isTransientTransportError(err)
|
||||
}
|
||||
|
||||
// cooldownWaitJitterCap bounds the random jitter added to cooldown waits so a
|
||||
// long wait is never extended by more than this amount.
|
||||
const cooldownWaitJitterCap = 2 * time.Second
|
||||
|
||||
273
sdk/cliproxy/auth/conductor_transport_retry_test.go
Normal file
273
sdk/cliproxy/auth/conductor_transport_retry_test.go
Normal file
@@ -0,0 +1,273 @@
|
||||
package auth
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/x509"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"sync"
|
||||
"syscall"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
internalconfig "github.com/router-for-me/CLIProxyAPI/v7/internal/config"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/registry"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executionregistry"
|
||||
cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor"
|
||||
)
|
||||
|
||||
func windowsCodexTLSHandshakeError() error {
|
||||
return &url.Error{
|
||||
Op: "Post",
|
||||
URL: "https://chatgpt.com/backend-api/codex/responses",
|
||||
Err: fmt.Errorf("tls: TLS handshake: %w", &net.OpError{
|
||||
Op: "read",
|
||||
Net: "tcp",
|
||||
Source: &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 1},
|
||||
Addr: &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 2},
|
||||
Err: errors.New("wsarecv: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond."),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
func dialRefusedError() error {
|
||||
return &url.Error{
|
||||
Op: "Post",
|
||||
URL: "https://chatgpt.com/backend-api/codex/responses",
|
||||
Err: &net.OpError{Op: "dial", Net: "tcp", Err: syscall.ECONNREFUSED},
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_ShouldRetryAfterError_RetriesPreHTTPTransportFailure(t *testing.T) {
|
||||
m := NewManager(nil, nil, nil)
|
||||
m.SetRetryConfig(1, 0, 0)
|
||||
model := "gpt-transport-retry-" + uuid.NewString()
|
||||
authID := "transport-retry-" + uuid.NewString()
|
||||
registry.GetGlobalRegistry().RegisterClient(authID, "codex", []*registry.ModelInfo{{ID: model}})
|
||||
t.Cleanup(func() { registry.GetGlobalRegistry().UnregisterClient(authID) })
|
||||
if _, errRegister := m.Register(context.Background(), &Auth{ID: authID, Provider: "codex"}); errRegister != nil {
|
||||
t.Fatalf("register auth: %v", errRegister)
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{name: "windows tls handshake", err: windowsCodexTLSHandshakeError(), want: true},
|
||||
{name: "dial refused", err: dialRefusedError(), want: true},
|
||||
{name: "unexpected eof", err: io.ErrUnexpectedEOF, want: true},
|
||||
{name: "unauthorized", err: &Error{HTTPStatus: http.StatusUnauthorized, Message: "unauthorized"}, want: false},
|
||||
{name: "canceled", err: context.Canceled, want: false},
|
||||
{name: "certificate", err: &url.Error{Op: "Post", URL: "https://chatgpt.com/backend-api/codex/responses", Err: x509.UnknownAuthorityError{}}, want: false},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
wait, shouldRetry := m.shouldRetryAfterError(tc.err, 0, []string{"codex"}, model, 0)
|
||||
if shouldRetry != tc.want {
|
||||
t.Fatalf("shouldRetryAfterError() = (%v, %t), want retry %t", wait, shouldRetry, tc.want)
|
||||
}
|
||||
if tc.want && wait != 0 {
|
||||
t.Fatalf("shouldRetryAfterError() wait = %v, want 0", wait)
|
||||
}
|
||||
if tc.want {
|
||||
if _, shouldRetry = m.shouldRetryAfterError(tc.err, 1, []string{"codex"}, model, 0); shouldRetry {
|
||||
t.Fatal("transport retried after the configured additional round")
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_MarkResult_PreHTTPTransportFailureDoesNotCooldown(t *testing.T) {
|
||||
previous := quotaCooldownDisabled.Load()
|
||||
quotaCooldownDisabled.Store(false)
|
||||
t.Cleanup(func() { quotaCooldownDisabled.Store(previous) })
|
||||
|
||||
prevTransient := transientErrorCooldownSeconds.Load()
|
||||
SetTransientErrorCooldownSeconds(5)
|
||||
t.Cleanup(func() { transientErrorCooldownSeconds.Store(prevTransient) })
|
||||
|
||||
model := "gpt-6-astra"
|
||||
cases := []struct {
|
||||
name string
|
||||
err *Error
|
||||
}{
|
||||
{name: "typed tls handshake", err: resultErrorFromError(windowsCodexTLSHandshakeError())},
|
||||
{name: "connection reset message", err: &Error{Message: "connection reset"}},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
m := NewManager(nil, nil, nil)
|
||||
auth := &Auth{ID: "auth-transport-" + uuid.NewString(), Provider: "codex"}
|
||||
if _, errRegister := m.Register(context.Background(), auth); errRegister != nil {
|
||||
t.Fatalf("register auth: %v", errRegister)
|
||||
}
|
||||
m.MarkResult(context.Background(), Result{
|
||||
AuthID: auth.ID,
|
||||
Provider: auth.Provider,
|
||||
Model: model,
|
||||
Success: false,
|
||||
Error: tc.err,
|
||||
})
|
||||
assertNoCooldown(t, m, auth.ID, model)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteRetriesPreHTTPTransportFailureWithoutCooling(t *testing.T) {
|
||||
previous := quotaCooldownDisabled.Load()
|
||||
quotaCooldownDisabled.Store(false)
|
||||
t.Cleanup(func() { quotaCooldownDisabled.Store(previous) })
|
||||
|
||||
manager := NewManager(nil, nil, nil)
|
||||
manager.SetRetryConfig(1, 0, 0)
|
||||
executor := &transportThenSuccessExecutor{
|
||||
identifier: "codex",
|
||||
fail: windowsCodexTLSHandshakeError(),
|
||||
}
|
||||
manager.RegisterExecutor(executor)
|
||||
|
||||
model := "gpt-6-astra-" + uuid.NewString()
|
||||
authID := "codex-transport-" + uuid.NewString()
|
||||
registry.GetGlobalRegistry().RegisterClient(authID, "codex", []*registry.ModelInfo{{ID: model}})
|
||||
t.Cleanup(func() { registry.GetGlobalRegistry().UnregisterClient(authID) })
|
||||
if _, errRegister := manager.Register(context.Background(), &Auth{ID: authID, Provider: "codex"}); errRegister != nil {
|
||||
t.Fatalf("register auth: %v", errRegister)
|
||||
}
|
||||
|
||||
resp, errExecute := manager.Execute(context.Background(), []string{"codex"}, cliproxyexecutor.Request{Model: model}, cliproxyexecutor.Options{})
|
||||
if errExecute != nil {
|
||||
t.Fatalf("Execute() error = %v, want success after transport retry", errExecute)
|
||||
}
|
||||
if string(resp.Payload) != "ok" {
|
||||
t.Fatalf("Execute() payload = %q, want %q", resp.Payload, "ok")
|
||||
}
|
||||
if calls := executor.callCount(); calls != 2 {
|
||||
t.Fatalf("executor calls = %d, want 2", calls)
|
||||
}
|
||||
assertNoCooldown(t, manager, authID, model)
|
||||
}
|
||||
|
||||
func TestExecuteDoesNotPoisonCredentialOnPreHTTPTransportFailure(t *testing.T) {
|
||||
previous := quotaCooldownDisabled.Load()
|
||||
quotaCooldownDisabled.Store(false)
|
||||
t.Cleanup(func() { quotaCooldownDisabled.Store(previous) })
|
||||
|
||||
manager := NewManager(nil, nil, nil)
|
||||
manager.SetRetryConfig(0, 0, 0)
|
||||
executor := &transportThenSuccessExecutor{
|
||||
identifier: "codex",
|
||||
fail: windowsCodexTLSHandshakeError(),
|
||||
}
|
||||
manager.RegisterExecutor(executor)
|
||||
|
||||
model := "gpt-6-astra-" + uuid.NewString()
|
||||
authID := "codex-transport-poison-" + uuid.NewString()
|
||||
registry.GetGlobalRegistry().RegisterClient(authID, "codex", []*registry.ModelInfo{{ID: model}})
|
||||
t.Cleanup(func() { registry.GetGlobalRegistry().UnregisterClient(authID) })
|
||||
if _, errRegister := manager.Register(context.Background(), &Auth{ID: authID, Provider: "codex"}); errRegister != nil {
|
||||
t.Fatalf("register auth: %v", errRegister)
|
||||
}
|
||||
|
||||
req := cliproxyexecutor.Request{Model: model}
|
||||
if _, errExecute := manager.Execute(context.Background(), []string{"codex"}, req, cliproxyexecutor.Options{}); errExecute == nil {
|
||||
t.Fatal("first Execute() error = nil, want transport failure")
|
||||
} else if shouldRetrySchedulerPick(errExecute) {
|
||||
t.Fatalf("first Execute() = %v, want raw transport error rather than auth_unavailable", errExecute)
|
||||
}
|
||||
assertNoCooldown(t, manager, authID, model)
|
||||
|
||||
resp, errExecute := manager.Execute(context.Background(), []string{"codex"}, req, cliproxyexecutor.Options{})
|
||||
if errExecute != nil {
|
||||
t.Fatalf("second Execute() error = %v, want success on the still-available credential", errExecute)
|
||||
}
|
||||
if string(resp.Payload) != "ok" {
|
||||
t.Fatalf("second Execute() payload = %q, want %q", resp.Payload, "ok")
|
||||
}
|
||||
if calls := executor.callCount(); calls != 2 {
|
||||
t.Fatalf("executor calls = %d, want 2", calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHomeExecuteRetriesPreHTTPTransportFailure(t *testing.T) {
|
||||
dispatcher := &retryContractHomeDispatcher{authIDs: []string{"home-retry-a"}}
|
||||
executor := &transportThenSuccessExecutor{
|
||||
identifier: "home-retry-contract",
|
||||
fail: windowsCodexTLSHandshakeError(),
|
||||
}
|
||||
manager := NewManager(nil, nil, nil)
|
||||
manager.SetConfig(&internalconfig.Config{Home: internalconfig.HomeConfig{Enabled: true}})
|
||||
manager.SetRetryConfig(1, 0, 0)
|
||||
manager.PublishHomeDispatch(dispatcher, executionregistry.New(), 1)
|
||||
manager.RegisterExecutor(executor)
|
||||
|
||||
resp, errExecute := manager.Execute(context.Background(), []string{"home-retry-contract"}, cliproxyexecutor.Request{Model: "gpt"}, cliproxyexecutor.Options{})
|
||||
if errExecute != nil {
|
||||
t.Fatalf("Execute() error = %v, want success after Home transport retry", errExecute)
|
||||
}
|
||||
if string(resp.Payload) != "ok" {
|
||||
t.Fatalf("Execute() payload = %q, want %q", resp.Payload, "ok")
|
||||
}
|
||||
if calls := executor.callCount(); calls != 2 {
|
||||
t.Fatalf("executor calls = %d, want 2", calls)
|
||||
}
|
||||
}
|
||||
|
||||
type transportThenSuccessExecutor struct {
|
||||
identifier string
|
||||
fail error
|
||||
|
||||
mu sync.Mutex
|
||||
calls int
|
||||
}
|
||||
|
||||
func (e *transportThenSuccessExecutor) Identifier() string { return e.identifier }
|
||||
|
||||
func (e *transportThenSuccessExecutor) Execute(context.Context, *Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) (cliproxyexecutor.Response, error) {
|
||||
if e.recordCall() == 1 {
|
||||
return cliproxyexecutor.Response{}, e.fail
|
||||
}
|
||||
return cliproxyexecutor.Response{Payload: []byte("ok")}, nil
|
||||
}
|
||||
|
||||
func (e *transportThenSuccessExecutor) ExecuteStream(context.Context, *Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) (*cliproxyexecutor.StreamResult, error) {
|
||||
if e.recordCall() == 1 {
|
||||
return nil, e.fail
|
||||
}
|
||||
chunks := make(chan cliproxyexecutor.StreamChunk)
|
||||
close(chunks)
|
||||
return &cliproxyexecutor.StreamResult{Chunks: chunks}, nil
|
||||
}
|
||||
|
||||
func (*transportThenSuccessExecutor) Refresh(_ context.Context, auth *Auth) (*Auth, error) {
|
||||
return auth, nil
|
||||
}
|
||||
|
||||
func (e *transportThenSuccessExecutor) CountTokens(context.Context, *Auth, cliproxyexecutor.Request, cliproxyexecutor.Options) (cliproxyexecutor.Response, error) {
|
||||
if e.recordCall() == 1 {
|
||||
return cliproxyexecutor.Response{}, e.fail
|
||||
}
|
||||
return cliproxyexecutor.Response{Payload: []byte("ok")}, nil
|
||||
}
|
||||
|
||||
func (*transportThenSuccessExecutor) HttpRequest(context.Context, *Auth, *http.Request) (*http.Response, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (e *transportThenSuccessExecutor) recordCall() int {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
e.calls++
|
||||
return e.calls
|
||||
}
|
||||
|
||||
func (e *transportThenSuccessExecutor) callCount() int {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
return e.calls
|
||||
}
|
||||
@@ -167,7 +167,7 @@ func TestRecoverableUnknownFailuresHaveFiniteCooldown(t *testing.T) {
|
||||
resultErr *Error
|
||||
}{
|
||||
{name: "model failure without error details", model: "gpt-5"},
|
||||
{name: "auth transport failure without status", resultErr: &Error{Message: "connection reset"}},
|
||||
{name: "auth failure without status or transport signature", resultErr: &Error{Message: "upstream exploded"}},
|
||||
}
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
|
||||
@@ -19,6 +19,12 @@ const ErrorCodeConnectionLifecycle = "connection_lifecycle"
|
||||
|
||||
const connectionLifecycleErrorCode = ErrorCodeConnectionLifecycle
|
||||
|
||||
// ErrorCodeTransientTransport marks pre-HTTP dial/TLS/DNS/reset failures that
|
||||
// must skip credential cooldown and remain eligible for request-retry rounds.
|
||||
const ErrorCodeTransientTransport = "transient_transport"
|
||||
|
||||
const transientTransportErrorCode = ErrorCodeTransientTransport
|
||||
|
||||
// ErrorCodeForceCooldown marks failures that must enforce credential cooldown.
|
||||
const ErrorCodeForceCooldown = "force_cooldown"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user