fix: fail over Codex manifest accounts

backend/internal/handler/openai_codex_models_handler.go

@@ -32,33 +32,54 @@ func (h *OpenAIGatewayHandler) CodexModels(c *gin.Context) {
 		return
 	}
 
-	account, err := h.gatewayService.SelectAccountForModel(c.Request.Context(), apiKey.GroupID, "", "")
-	if err != nil {
+	maxAccountSwitches := h.maxAccountSwitches
+	if maxAccountSwitches <= 0 {
+		maxAccountSwitches = 3
+	}
+	failedAccountIDs := make(map[int64]struct{})
+	switchCount := 0
+	var lastUpstreamErr error
+
+	for {
+		account, err := h.gatewayService.SelectAccountForModelWithExclusions(c.Request.Context(), apiKey.GroupID, "", "", failedAccountIDs)
+		if err != nil {
+			if c.Request.Context().Err() != nil {
+				return
+			}
+			if lastUpstreamErr != nil {
+				h.errorResponse(c, infraerrors.Code(lastUpstreamErr), "upstream_error", infraerrors.Message(lastUpstreamErr))
+				return
+			}
+			h.errorResponse(c, http.StatusServiceUnavailable, "upstream_error", "No available OpenAI accounts")
+			return
+		}
+
+		manifest, err := h.gatewayService.FetchCodexModelsManifest(c.Request.Context(), account, c.Query("client_version"), c.GetHeader("If-None-Match"))
+		if err != nil {
+			if c.Request.Context().Err() != nil {
+				return
+			}
+			if service.IsRetryableCodexModelsManifestError(err) && switchCount < maxAccountSwitches {
+				failedAccountIDs[account.ID] = struct{}{}
+				switchCount++
+				lastUpstreamErr = err
+				continue
+			}
+			h.errorResponse(c, infraerrors.Code(err), "upstream_error", infraerrors.Message(err))
+			return
+		}
 		if c.Request.Context().Err() != nil {
 			return
 		}
-		h.errorResponse(c, http.StatusServiceUnavailable, "upstream_error", "No available OpenAI accounts")
-		return
-	}
 
-	manifest, err := h.gatewayService.FetchCodexModelsManifest(c.Request.Context(), account, c.Query("client_version"), c.GetHeader("If-None-Match"))
-	if err != nil {
-		if c.Request.Context().Err() != nil {
+		if manifest.ETag != "" {
+			c.Header("ETag", manifest.ETag)
+		}
+		if manifest.NotModified {
+			c.Status(http.StatusNotModified)
 			return
 		}
-		h.errorResponse(c, infraerrors.Code(err), "upstream_error", infraerrors.Message(err))
+		c.Data(http.StatusOK, "application/json", manifest.Body)
 		return
 	}
-	if c.Request.Context().Err() != nil {
-		return
-	}
-
-	if manifest.ETag != "" {
-		c.Header("ETag", manifest.ETag)
-	}
-	if manifest.NotModified {
-		c.Status(http.StatusNotModified)
-		return
-	}
-	c.Data(http.StatusOK, "application/json", manifest.Body)
 }

backend/internal/handler/openai_codex_models_handler_test.go

@@ -2,13 +2,95 @@ package handler
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
+	"strings"
+	"sync"
 	"testing"
 
+	"github.com/Wei-Shaw/sub2api/internal/config"
+	middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
+	"github.com/Wei-Shaw/sub2api/internal/service"
 	"github.com/gin-gonic/gin"
 )
 
+type codexModelsFailoverAccountRepo struct {
+	service.AccountRepository
+	accounts []service.Account
+}
+
+func (r codexModelsFailoverAccountRepo) GetByID(_ context.Context, id int64) (*service.Account, error) {
+	for i := range r.accounts {
+		if r.accounts[i].ID == id {
+			account := r.accounts[i]
+			return &account, nil
+		}
+	}
+	return nil, service.ErrNoAvailableAccounts
+}
+
+func (r codexModelsFailoverAccountRepo) ListSchedulableByPlatform(_ context.Context, platform string) ([]service.Account, error) {
+	accounts := make([]service.Account, 0, len(r.accounts))
+	for _, account := range r.accounts {
+		if account.Platform == platform {
+			accounts = append(accounts, account)
+		}
+	}
+	return accounts, nil
+}
+
+type codexModelsFailoverHTTPUpstream struct {
+	service.HTTPUpstream
+	mu          sync.Mutex
+	accountIDs  []int64
+	firstErr    error
+	firstStatus int
+	statuses    map[int64]int
+}
+
+func (u *codexModelsFailoverHTTPUpstream) Do(_ *http.Request, _ string, accountID int64, _ int) (*http.Response, error) {
+	u.mu.Lock()
+	u.accountIDs = append(u.accountIDs, accountID)
+	u.mu.Unlock()
+
+	status, hasStatus := u.statuses[accountID]
+	if accountID == 1 || hasStatus {
+		if u.firstErr != nil {
+			return nil, u.firstErr
+		}
+		if !hasStatus {
+			status = u.firstStatus
+		}
+		if status == 0 {
+			status = http.StatusServiceUnavailable
+		}
+		return &http.Response{
+			StatusCode: status,
+			Status:     http.StatusText(status),
+			Header:     make(http.Header),
+			Body: io.NopCloser(strings.NewReader(
+				`{"error":{"message":"No available OpenAI accounts","type":"upstream_error"}}`,
+			)),
+		}, nil
+	}
+	return &http.Response{
+		StatusCode: http.StatusOK,
+		Status:     "200 OK",
+		Header:     make(http.Header),
+		Body:       io.NopCloser(strings.NewReader(`{"models":[{"slug":"gpt-5.6-sol"}]}`)),
+	}, nil
+}
+
+func (u *codexModelsFailoverHTTPUpstream) calls() []int64 {
+	u.mu.Lock()
+	defer u.mu.Unlock()
+	return append([]int64(nil), u.accountIDs...)
+}
+
 func TestCodexModelsCanceledRequestDoesNotWriteResponse(t *testing.T) {
 	gin.SetMode(gin.TestMode)
 	recorder := httptest.NewRecorder()
@@ -24,3 +106,183 @@ func TestCodexModelsCanceledRequestDoesNotWriteResponse(t *testing.T) {
 		t.Fatalf("canceled request wrote an HTTP response: status=%d body=%q", recorder.Code, recorder.Body.String())
 	}
 }
+
+func TestCodexModelsFailsOverFromRetryableUpstreamStatus(t *testing.T) {
+	retryableStatuses := []int{
+		http.StatusTooManyRequests,
+		http.StatusInternalServerError,
+		http.StatusBadGateway,
+		http.StatusServiceUnavailable,
+		http.StatusGatewayTimeout,
+	}
+	for _, status := range retryableStatuses {
+		t.Run(http.StatusText(status), func(t *testing.T) {
+			handler, upstream, groupID := newCodexModelsFailoverTestHandler(status)
+			recorder := performCodexModelsRequest(t, handler, groupID)
+
+			if got, want := upstream.calls(), []int64{1, 2}; !equalInt64Slices(got, want) {
+				t.Fatalf("upstream account calls: got %v, want %v", got, want)
+			}
+			if recorder.Code != http.StatusOK {
+				t.Fatalf("status: got %d, want %d; body=%s", recorder.Code, http.StatusOK, recorder.Body.String())
+			}
+			if got, want := recorder.Body.String(), `{"models":[{"slug":"gpt-5.6-sol"}]}`; got != want {
+				t.Fatalf("body: got %q, want %q", got, want)
+			}
+		})
+	}
+}
+
+func TestCodexModelsFailsOverFromUpstreamTransportError(t *testing.T) {
+	handler, upstream, groupID := newCodexModelsFailoverTestHandler(http.StatusServiceUnavailable)
+	upstream.firstErr = &net.OpError{
+		Op:  "read",
+		Net: "tcp",
+		Err: errors.New("connection reset"),
+	}
+	recorder := performCodexModelsRequest(t, handler, groupID)
+
+	if got, want := upstream.calls(), []int64{1, 2}; !equalInt64Slices(got, want) {
+		t.Fatalf("upstream account calls: got %v, want %v", got, want)
+	}
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status: got %d, want %d; body=%s", recorder.Code, http.StatusOK, recorder.Body.String())
+	}
+}
+
+func TestCodexModelsDoesNotFailOverFromPermanentUpstreamStatus(t *testing.T) {
+	statuses := []int{
+		http.StatusBadRequest,
+		http.StatusUnauthorized,
+		http.StatusForbidden,
+		http.StatusNotFound,
+		600,
+	}
+	for _, status := range statuses {
+		t.Run(fmt.Sprintf("status_%d", status), func(t *testing.T) {
+			handler, upstream, groupID := newCodexModelsFailoverTestHandler(status)
+			recorder := performCodexModelsRequest(t, handler, groupID)
+
+			if got, want := upstream.calls(), []int64{1}; !equalInt64Slices(got, want) {
+				t.Fatalf("upstream account calls: got %v, want %v", got, want)
+			}
+			if recorder.Code != http.StatusBadGateway {
+				t.Fatalf("status: got %d, want %d; body=%s", recorder.Code, http.StatusBadGateway, recorder.Body.String())
+			}
+		})
+	}
+}
+
+func TestCodexModelsDoesNotFailOverFromUpstreamConfigurationError(t *testing.T) {
+	handler, upstream, groupID := newCodexModelsFailoverTestHandler(http.StatusServiceUnavailable)
+	upstream.firstErr = errors.New("invalid proxy URL")
+	recorder := performCodexModelsRequest(t, handler, groupID)
+
+	if got, want := upstream.calls(), []int64{1}; !equalInt64Slices(got, want) {
+		t.Fatalf("upstream account calls: got %v, want %v", got, want)
+	}
+	if recorder.Code != http.StatusBadGateway {
+		t.Fatalf("status: got %d, want %d; body=%s", recorder.Code, http.StatusBadGateway, recorder.Body.String())
+	}
+}
+
+func TestCodexModelsReturnsLastUpstreamErrorWhenAccountsAreExhausted(t *testing.T) {
+	handler, upstream, groupID := newCodexModelsFailoverTestHandler(http.StatusServiceUnavailable)
+	upstream.statuses = map[int64]int{
+		1: http.StatusServiceUnavailable,
+		2: http.StatusGatewayTimeout,
+	}
+	recorder := performCodexModelsRequest(t, handler, groupID)
+
+	if got, want := upstream.calls(), []int64{1, 2}; !equalInt64Slices(got, want) {
+		t.Fatalf("upstream account calls: got %v, want %v", got, want)
+	}
+	if recorder.Code != http.StatusBadGateway {
+		t.Fatalf("status: got %d, want %d; body=%s", recorder.Code, http.StatusBadGateway, recorder.Body.String())
+	}
+	if body := recorder.Body.String(); !strings.Contains(body, "upstream error 504") {
+		t.Fatalf("body does not preserve the last upstream error: %s", body)
+	}
+}
+
+func TestCodexModelsHonorsAccountSwitchLimit(t *testing.T) {
+	handler, upstream, groupID := newCodexModelsFailoverTestHandlerWithAccountCount(http.StatusServiceUnavailable, 4, 2)
+	upstream.statuses = map[int64]int{
+		1: http.StatusServiceUnavailable,
+		2: http.StatusBadGateway,
+		3: http.StatusGatewayTimeout,
+		4: http.StatusInternalServerError,
+	}
+	recorder := performCodexModelsRequest(t, handler, groupID)
+
+	if got, want := upstream.calls(), []int64{1, 2, 3}; !equalInt64Slices(got, want) {
+		t.Fatalf("upstream account calls: got %v, want %v", got, want)
+	}
+	if recorder.Code != http.StatusBadGateway {
+		t.Fatalf("status: got %d, want %d; body=%s", recorder.Code, http.StatusBadGateway, recorder.Body.String())
+	}
+	if body := recorder.Body.String(); !strings.Contains(body, "upstream error 504") {
+		t.Fatalf("body does not preserve the limit-ending upstream error: %s", body)
+	}
+}
+
+func newCodexModelsFailoverTestHandler(firstStatus int) (*OpenAIGatewayHandler, *codexModelsFailoverHTTPUpstream, int64) {
+	return newCodexModelsFailoverTestHandlerWithAccountCount(firstStatus, 2, 3)
+}
+
+func newCodexModelsFailoverTestHandlerWithAccountCount(firstStatus, accountCount, maxSwitches int) (*OpenAIGatewayHandler, *codexModelsFailoverHTTPUpstream, int64) {
+	gin.SetMode(gin.TestMode)
+	groupID := int64(42)
+	accounts := make([]service.Account, 0, accountCount)
+	for i := 1; i <= accountCount; i++ {
+		accounts = append(accounts, service.Account{
+			ID:          int64(i),
+			Name:        fmt.Sprintf("upstream-%d", i),
+			Platform:    service.PlatformOpenAI,
+			Type:        service.AccountTypeAPIKey,
+			Status:      service.StatusActive,
+			Schedulable: true,
+			Priority:    i - 1,
+			Concurrency: 1,
+			Credentials: map[string]any{
+				"api_key":  fmt.Sprintf("sk-%d", i),
+				"base_url": fmt.Sprintf("https://upstream-%d.example/v1", i),
+			},
+		})
+	}
+	upstream := &codexModelsFailoverHTTPUpstream{firstStatus: firstStatus}
+	cfg := &config.Config{RunMode: config.RunModeSimple}
+	gatewayService := service.NewOpenAIGatewayService(
+		codexModelsFailoverAccountRepo{accounts: accounts},
+		nil, nil, nil, nil, nil, nil, cfg, nil, nil, nil, nil, nil,
+		upstream,
+		nil, nil, nil, nil, nil, nil, nil, nil,
+	)
+	return &OpenAIGatewayHandler{gatewayService: gatewayService, maxAccountSwitches: maxSwitches}, upstream, groupID
+}
+
+func performCodexModelsRequest(t *testing.T, handler *OpenAIGatewayHandler, groupID int64) *httptest.ResponseRecorder {
+	t.Helper()
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodGet, "/v1/models?client_version=0.144.0", nil)
+	c.Set(string(middleware2.ContextKeyAPIKey), &service.APIKey{
+		GroupID: &groupID,
+		Group:   &service.Group{ID: groupID, Platform: service.PlatformOpenAI},
+	})
+
+	handler.CodexModels(c)
+	return recorder
+}
+
+func equalInt64Slices(got, want []int64) bool {
+	if len(got) != len(want) {
+		return false
+	}
+	for i := range got {
+		if got[i] != want[i] {
+			return false
+		}
+	}
+	return true
+}

backend/internal/service/openai_codex_models_service.go

@@ -3,8 +3,10 @@ package service
 import (
 	"context"
 	"crypto/sha256"
+	"errors"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/url"
 	"sort"
@@ -14,6 +16,7 @@ import (
 
 	infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
 	"github.com/Wei-Shaw/sub2api/internal/pkg/httpclient"
+	"golang.org/x/net/http2"
 	"golang.org/x/sync/singleflight"
 )
 
@@ -38,6 +41,94 @@ type CodexModelsManifest struct {
 	NotModified bool
 }
 
+type codexModelsManifestUpstreamError struct {
+	err       error
+	retryable bool
+}
+
+func (e *codexModelsManifestUpstreamError) Error() string { return e.err.Error() }
+
+func (e *codexModelsManifestUpstreamError) Unwrap() error { return e.err }
+
+// IsRetryableCodexModelsManifestError reports whether another selected account
+// may succeed without changing the request. Configuration and upstream 4xx
+// responses, except 429, are intentionally not retried.
+func IsRetryableCodexModelsManifestError(err error) bool {
+	var upstreamErr *codexModelsManifestUpstreamError
+	return errors.As(err, &upstreamErr) && upstreamErr.retryable
+}
+
+func isRetryableCodexModelsManifestTransportError(err error) bool {
+	if err == nil || errors.Is(err, context.Canceled) {
+		return false
+	}
+	if errors.Is(err, context.DeadlineExceeded) ||
+		errors.Is(err, io.EOF) ||
+		errors.Is(err, io.ErrUnexpectedEOF) ||
+		errors.Is(err, net.ErrClosed) {
+		return true
+	}
+
+	var opErr *net.OpError
+	if errors.As(err, &opErr) {
+		return true
+	}
+	var dnsErr *net.DNSError
+	if errors.As(err, &dnsErr) {
+		return true
+	}
+	var goAwayErr http2.GoAwayError
+	if errors.As(err, &goAwayErr) {
+		return true
+	}
+	var streamErr http2.StreamError
+	if errors.As(err, &streamErr) {
+		return true
+	}
+	var connectionErr http2.ConnectionError
+	if errors.As(err, &connectionErr) {
+		return true
+	}
+	var netErr net.Error
+	if errors.As(err, &netErr) && netErr.Timeout() {
+		return true
+	}
+
+	// net/http uses unexported HTTP/2 error types, so typed matching is not
+	// possible for errors produced by the standard library transport.
+	message := strings.ToLower(err.Error())
+	if strings.Contains(message, "http2:") &&
+		(strings.Contains(message, "goaway") ||
+			strings.Contains(message, "refused_stream") ||
+			strings.Contains(message, "frame too large")) {
+		return true
+	}
+	if strings.Contains(message, "stream error: stream id ") {
+		return true
+	}
+	for _, code := range []http2.ErrCode{
+		http2.ErrCodeNo,
+		http2.ErrCodeProtocol,
+		http2.ErrCodeInternal,
+		http2.ErrCodeFlowControl,
+		http2.ErrCodeSettingsTimeout,
+		http2.ErrCodeStreamClosed,
+		http2.ErrCodeFrameSize,
+		http2.ErrCodeRefusedStream,
+		http2.ErrCodeCancel,
+		http2.ErrCodeCompression,
+		http2.ErrCodeConnect,
+		http2.ErrCodeEnhanceYourCalm,
+		http2.ErrCodeInadequateSecurity,
+		http2.ErrCodeHTTP11Required,
+	} {
+		if strings.Contains(message, "connection error: "+strings.ToLower(code.String())) {
+			return true
+		}
+	}
+	return false
+}
+
 type codexModelsManifestRequest struct {
 	url                 string
 	headers             http.Header
@@ -298,7 +389,10 @@ func (s *OpenAIGatewayService) fetchCodexModelsManifestUpstream(ctx context.Cont
 		resp, err = client.Do(req)
 	}
 	if err != nil {
-		return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_UPSTREAM_FAILED", "codex models manifest request failed: %v", err)
+		return nil, &codexModelsManifestUpstreamError{
+			err:       infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_UPSTREAM_FAILED", "codex models manifest request failed: %v", err),
+			retryable: isRetryableCodexModelsManifestTransportError(err),
+		}
 	}
 	defer func() { _ = resp.Body.Close() }()
 
@@ -311,12 +405,19 @@ func (s *OpenAIGatewayService) fetchCodexModelsManifestUpstream(ctx context.Cont
 		if message == "" {
 			message = resp.Status
 		}
-		return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_UPSTREAM_FAILED", "codex models manifest upstream error %d: %s", resp.StatusCode, message)
+		return nil, &codexModelsManifestUpstreamError{
+			err: infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_UPSTREAM_FAILED", "codex models manifest upstream error %d: %s", resp.StatusCode, message),
+			retryable: resp.StatusCode == http.StatusTooManyRequests ||
+				(resp.StatusCode >= http.StatusInternalServerError && resp.StatusCode < 600),
+		}
 	}
 
 	body, err := io.ReadAll(io.LimitReader(resp.Body, codexModelsManifestBodyLimit))
 	if err != nil {
-		return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_UPSTREAM_FAILED", "read codex models manifest response: %v", err)
+		return nil, &codexModelsManifestUpstreamError{
+			err:       infraerrors.Newf(http.StatusBadGateway, "OPENAI_CODEX_MODELS_UPSTREAM_FAILED", "read codex models manifest response: %v", err),
+			retryable: isRetryableCodexModelsManifestTransportError(err),
+		}
 	}
 	return &CodexModelsManifest{Body: body, ETag: resp.Header.Get("ETag")}, nil
 }

backend/internal/service/openai_codex_models_service_test.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -15,6 +17,7 @@ import (
 	"github.com/Wei-Shaw/sub2api/internal/config"
 	infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
 	"github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
+	"golang.org/x/net/http2"
 )
 
 type codexModelsHTTPUpstreamStub struct {
@@ -49,6 +52,73 @@ func (s *codexModelsHTTPUpstreamStub) DoWithTLS(req *http.Request, proxyURL stri
 	return s.Do(req, proxyURL, accountID, accountConcurrency)
 }
 
+func TestIsRetryableCodexModelsManifestTransportError(t *testing.T) {
+	tests := []struct {
+		name      string
+		err       error
+		retryable bool
+	}{
+		{name: "nil", err: nil},
+		{name: "configuration error", err: errors.New("invalid proxy URL")},
+		{name: "upstream configuration error", err: errors.New("upstream error: invalid proxy")},
+		{name: "proxy connection configuration error", err: errors.New("proxy connection error: invalid configuration")},
+		{name: "canceled request", err: context.Canceled},
+		{
+			name: "redirect policy error",
+			err: &url.Error{
+				Op:  "Get",
+				URL: "https://upstream.example/v1/models",
+				Err: errors.New("stopped after 10 redirects"),
+			},
+		},
+		{name: "deadline exceeded", err: context.DeadlineExceeded, retryable: true},
+		{name: "unexpected EOF", err: io.ErrUnexpectedEOF, retryable: true},
+		{name: "closed connection", err: net.ErrClosed, retryable: true},
+		{
+			name: "network operation",
+			err: &net.OpError{
+				Op:  "read",
+				Net: "tcp",
+				Err: errors.New("connection reset"),
+			},
+			retryable: true,
+		},
+		{
+			name:      "DNS error",
+			err:       &net.DNSError{Err: "temporary failure", Name: "upstream.example"},
+			retryable: true,
+		},
+		{
+			name:      "typed HTTP2 GOAWAY",
+			err:       http2.GoAwayError{ErrCode: http2.ErrCodeNo},
+			retryable: true,
+		},
+		{
+			name:      "stdlib HTTP2 GOAWAY",
+			err:       errors.New("http2: server sent GOAWAY and closed the connection; LastStreamID=1, ErrCode=NO_ERROR"),
+			retryable: true,
+		},
+		{
+			name:      "stdlib HTTP2 refused stream",
+			err:       errors.New("stream error: stream ID 3; REFUSED_STREAM"),
+			retryable: true,
+		},
+		{
+			name:      "stdlib HTTP2 connection error",
+			err:       errors.New(`Get "https://upstream.example/v1/models": connection error: PROTOCOL_ERROR`),
+			retryable: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isRetryableCodexModelsManifestTransportError(tt.err); got != tt.retryable {
+				t.Fatalf("retryable = %v, want %v", got, tt.retryable)
+			}
+		})
+	}
+}
+
 func newCodexModelsAPIKeyTestService(upstream HTTPUpstream) *OpenAIGatewayService {
 	return &OpenAIGatewayService{
 		cfg: &config.Config{Security: config.SecurityConfig{URLAllowlist: config.URLAllowlistConfig{