diff --git a/sdk/cliproxy/auth/conductor_cooldown.go b/sdk/cliproxy/auth/conductor_cooldown.go index 28f88a139..22041fa59 100644 --- a/sdk/cliproxy/auth/conductor_cooldown.go +++ b/sdk/cliproxy/auth/conductor_cooldown.go @@ -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 diff --git a/sdk/cliproxy/auth/conductor_selection.go b/sdk/cliproxy/auth/conductor_selection.go index a03edd9dc..70ea9c52a 100644 --- a/sdk/cliproxy/auth/conductor_selection.go +++ b/sdk/cliproxy/auth/conductor_selection.go @@ -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 diff --git a/sdk/cliproxy/auth/conductor_transport_retry_test.go b/sdk/cliproxy/auth/conductor_transport_retry_test.go new file mode 100644 index 000000000..8b62778f4 --- /dev/null +++ b/sdk/cliproxy/auth/conductor_transport_retry_test.go @@ -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 +} diff --git a/sdk/cliproxy/auth/cooldown_backoff_test.go b/sdk/cliproxy/auth/cooldown_backoff_test.go index 73a7bdcf3..dccb40ce1 100644 --- a/sdk/cliproxy/auth/cooldown_backoff_test.go +++ b/sdk/cliproxy/auth/cooldown_backoff_test.go @@ -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) { diff --git a/sdk/cliproxy/auth/errors.go b/sdk/cliproxy/auth/errors.go index 7ae4f8581..206047620 100644 --- a/sdk/cliproxy/auth/errors.go +++ b/sdk/cliproxy/auth/errors.go @@ -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"