feat: OpenAI Team 联动熔断
OpenAI OAuth 账户收到 402 deactivated_workspace（ChatGPT Team 工作区被停用）时， 将同一 Team（credentials.chatgpt_account_id 相同）的其余 active 账户一并置为 error 并进程内立即熔断，让调度瞬间切走，避免流量继续打到已死的兄弟账户。 - 触发条件仅限 402 + detail.code == deactivated_workspace；通用 402 与 401 行为不变 - 调用点两处：fastpath 各类早退之前 + HandleUpstreamError 顶部，进程内 60s 按 Team 去重 - 逐账户 SetError（错误信息标注触发账户），单账户失败不中断其余 - 默认生效，无配置开关
This commit is contained in:
feeeei
@@ -72,6 +72,10 @@ func (s *OpenAIGatewayService) handleOpenAIAccountUpstreamError(ctx context.Cont
|
||||
if s == nil || account == nil {
|
||||
return false
|
||||
}
|
||||
// Team 联动熔断必须先于 model-not-found 与账户级临时不可调度规则的早退。
|
||||
if s.rateLimitService != nil {
|
||||
s.rateLimitService.maybeHandleOpenAITeamLinkedError(stateCtx, account, statusCode, responseBody)
|
||||
}
|
||||
stateCtx = withTempUnschedulableModel(stateCtx, canonicalModel)
|
||||
if s.rateLimitService != nil && len(canonicalModel) > 0 && s.rateLimitService.HandleUpstreamModelNotFound(stateCtx, account, canonicalModel[0], statusCode, responseBody) {
|
||||
return true
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/tidwall/gjson"
|
||||
)
|
||||
|
||||
const (
|
||||
openAITeamLinkedErrorDedupTTL = 60 * time.Second
|
||||
openAITeamLinkedErrorFanoutTimeout = 30 * time.Second
|
||||
openAITeamLinkedErrorBlockReason = "team_linked_error"
|
||||
)
|
||||
|
||||
// maybeHandleOpenAITeamLinkedError 在 OpenAI OAuth 账户收到 402 deactivated_workspace
|
||||
// （ChatGPT Team 工作区被停用）时，把同一 Team（credentials.chatgpt_account_id 相同）
|
||||
// 的其余 active 账户一并置为 error 并立即熔断。触发账户自身不在 fan-out 范围内，
|
||||
// 仍由常规 402 处理标记。
|
||||
func (s *RateLimitService) maybeHandleOpenAITeamLinkedError(ctx context.Context, account *Account, statusCode int, responseBody []byte) {
|
||||
if s == nil || s.accountRepo == nil || statusCode != http.StatusPaymentRequired || !isOpenAIOAuthAccount(account) {
|
||||
return
|
||||
}
|
||||
if gjson.GetBytes(responseBody, "detail.code").String() != "deactivated_workspace" {
|
||||
return
|
||||
}
|
||||
teamID := strings.TrimSpace(account.GetChatGPTAccountID())
|
||||
if teamID == "" {
|
||||
return
|
||||
}
|
||||
if !s.markOpenAITeamLinkedFired(teamID) {
|
||||
return
|
||||
}
|
||||
// 上游报错场景请求 ctx 往往已被取消，落库需要独立生命周期。
|
||||
opCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), openAITeamLinkedErrorFanoutTimeout)
|
||||
defer cancel()
|
||||
|
||||
accounts, err := s.accountRepo.ListByPlatform(opCtx, PlatformOpenAI)
|
||||
if err != nil {
|
||||
slog.Warn("openai_team_linked_error_list_failed", "trigger_account_id", account.ID, "error", err)
|
||||
return
|
||||
}
|
||||
var targets []*Account
|
||||
for i := range accounts {
|
||||
acc := &accounts[i]
|
||||
if acc.ID == account.ID || acc.IsShadow() || strings.TrimSpace(acc.GetChatGPTAccountID()) != teamID {
|
||||
continue
|
||||
}
|
||||
targets = append(targets, acc)
|
||||
}
|
||||
if len(targets) == 0 {
|
||||
return
|
||||
}
|
||||
// 先全部进程内熔断（微秒级生效），再逐个落库，避免后面的账户等待前面的 DB 写入。
|
||||
for _, acc := range targets {
|
||||
s.notifyAccountSchedulingBlocked(acc, time.Time{}, openAITeamLinkedErrorBlockReason)
|
||||
}
|
||||
errorMsg := fmt.Sprintf("Workspace deactivated (402): team-linked error triggered by account #%d", account.ID)
|
||||
marked := 0
|
||||
for _, acc := range targets {
|
||||
// 单账户写入失败不中断其余账户；进程内熔断已先行，且该账户仍为 active，
|
||||
// 下一个 402 在去重 TTL 过期后会重新触发 fan-out。
|
||||
if err := s.accountRepo.SetError(opCtx, acc.ID, errorMsg); err != nil {
|
||||
slog.Warn("openai_team_linked_error_set_error_failed", "account_id", acc.ID, "error", err)
|
||||
continue
|
||||
}
|
||||
marked++
|
||||
}
|
||||
slog.Warn("openai_team_linked_error_fanout",
|
||||
"trigger_account_id", account.ID,
|
||||
"chatgpt_account_id", teamID,
|
||||
"affected", marked,
|
||||
"targets", len(targets),
|
||||
)
|
||||
}
|
||||
|
||||
// markOpenAITeamLinkedFired 以 teamID 为键做进程内去重：TTL 内同一 Team 只允许一次 fan-out。
|
||||
func (s *RateLimitService) markOpenAITeamLinkedFired(teamID string) bool {
|
||||
now := time.Now()
|
||||
s.openaiTeamLinkedMu.Lock()
|
||||
defer s.openaiTeamLinkedMu.Unlock()
|
||||
if expiry, ok := s.openaiTeamLinkedRecent[teamID]; ok && expiry.After(now) {
|
||||
return false
|
||||
}
|
||||
if s.openaiTeamLinkedRecent == nil {
|
||||
s.openaiTeamLinkedRecent = make(map[string]time.Time)
|
||||
}
|
||||
for k, v := range s.openaiTeamLinkedRecent {
|
||||
if !v.After(now) {
|
||||
delete(s.openaiTeamLinkedRecent, k)
|
||||
}
|
||||
}
|
||||
s.openaiTeamLinkedRecent[teamID] = now.Add(openAITeamLinkedErrorDedupTTL)
|
||||
return true
|
||||
}
|
||||
@@ -32,6 +32,10 @@ type RateLimitService struct {
|
||||
runtimeBlocker AccountRuntimeBlocker
|
||||
usageCacheMu sync.RWMutex
|
||||
usageCache map[int64]*geminiUsageCacheEntry
|
||||
|
||||
// OpenAI Team 联动熔断的进程内去重：teamID → 去重窗口截止时间
|
||||
openaiTeamLinkedMu sync.Mutex
|
||||
openaiTeamLinkedRecent map[string]time.Time
|
||||
}
|
||||
|
||||
type AccountRuntimeBlocker interface {
|
||||
@@ -268,6 +272,9 @@ func (s *RateLimitService) CheckErrorPolicy(ctx context.Context, account *Accoun
|
||||
// 返回是否应该停止该账号的调度
|
||||
func (s *RateLimitService) HandleUpstreamError(ctx context.Context, account *Account, statusCode int, headers http.Header, responseBody []byte, requestedModel ...string) (shouldDisable bool) {
|
||||
ctx = withTempUnschedulableModel(ctx, requestedModel)
|
||||
// Team 联动熔断必须先于池模式/自定义错误码/临时不可调度的各类早退；
|
||||
// 同请求内与 fastpath 调用点的重复触发由方法内去重吸收。
|
||||
s.maybeHandleOpenAITeamLinkedError(ctx, account, statusCode, responseBody)
|
||||
customErrorCodesEnabled := account.IsCustomErrorCodesEnabled()
|
||||
|
||||
// 池模式默认不标记本地账号状态；但管理员显式配置的临时不可调度规则优先。
|
||||
|
||||
@@ -0,0 +1,192 @@
|
||||
//go:build unit
|
||||
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"testing"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
const teamLinkedDeactivatedBody = `{"detail":{"code":"deactivated_workspace","message":"This workspace has been deactivated."}}`
|
||||
|
||||
type teamLinkedAccountRepoStub struct {
|
||||
mockAccountRepoForGemini
|
||||
teamAccounts []Account
|
||||
listErr error
|
||||
listCalls int
|
||||
setErrorIDs []int64
|
||||
setErrorMsgs map[int64]string
|
||||
failSetError map[int64]error
|
||||
}
|
||||
|
||||
// ListByPlatform 镜像真实仓库语义：仅返回该平台的 active 账户。
|
||||
func (r *teamLinkedAccountRepoStub) ListByPlatform(ctx context.Context, platform string) ([]Account, error) {
|
||||
r.listCalls++
|
||||
if r.listErr != nil {
|
||||
return nil, r.listErr
|
||||
}
|
||||
out := make([]Account, 0, len(r.teamAccounts))
|
||||
for _, acc := range r.teamAccounts {
|
||||
if acc.Platform == platform && acc.Status == StatusActive {
|
||||
out = append(out, acc)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (r *teamLinkedAccountRepoStub) SetError(ctx context.Context, id int64, errorMsg string) error {
|
||||
if err, ok := r.failSetError[id]; ok {
|
||||
return err
|
||||
}
|
||||
r.setErrorIDs = append(r.setErrorIDs, id)
|
||||
if r.setErrorMsgs == nil {
|
||||
r.setErrorMsgs = make(map[int64]string)
|
||||
}
|
||||
r.setErrorMsgs[id] = errorMsg
|
||||
return nil
|
||||
}
|
||||
|
||||
func newTeamLinkedAccount(id int64, teamID string) Account {
|
||||
return Account{
|
||||
ID: id,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeOAuth,
|
||||
Status: StatusActive,
|
||||
Credentials: map[string]any{"chatgpt_account_id": teamID},
|
||||
}
|
||||
}
|
||||
|
||||
// newTeamLinkedFixture: #1 触发者(team-A) #2 同队 #3 异队 #4 apikey #5 影子 #6 同队 #7 同队但已 error
|
||||
func newTeamLinkedFixture() []Account {
|
||||
parentID := int64(1)
|
||||
shadow := Account{
|
||||
ID: 5,
|
||||
Platform: PlatformOpenAI,
|
||||
Type: AccountTypeOAuth,
|
||||
Status: StatusActive,
|
||||
ParentAccountID: &parentID,
|
||||
}
|
||||
apikey := newTeamLinkedAccount(4, "team-A")
|
||||
apikey.Type = AccountTypeAPIKey
|
||||
erroredSibling := newTeamLinkedAccount(7, "team-A")
|
||||
erroredSibling.Status = StatusError
|
||||
return []Account{
|
||||
newTeamLinkedAccount(1, "team-A"),
|
||||
newTeamLinkedAccount(2, "team-A"),
|
||||
newTeamLinkedAccount(3, "team-B"),
|
||||
apikey,
|
||||
shadow,
|
||||
newTeamLinkedAccount(6, "team-A"),
|
||||
erroredSibling,
|
||||
}
|
||||
}
|
||||
|
||||
func newTeamLinkedTestService(repo *teamLinkedAccountRepoStub) (*RateLimitService, *runtimeBlockRecorder) {
|
||||
rl := NewRateLimitService(repo, nil, &config.Config{}, nil, nil)
|
||||
blocker := &runtimeBlockRecorder{}
|
||||
rl.SetAccountRuntimeBlocker(blocker)
|
||||
return rl, blocker
|
||||
}
|
||||
|
||||
func TestTeamLinkedError_FanoutMarksSameTeamAccounts(t *testing.T) {
|
||||
repo := &teamLinkedAccountRepoStub{teamAccounts: newTeamLinkedFixture()}
|
||||
rl, blocker := newTeamLinkedTestService(repo)
|
||||
trigger := newTeamLinkedAccount(1, "team-A")
|
||||
|
||||
shouldDisable := rl.HandleUpstreamError(context.Background(), &trigger, http.StatusPaymentRequired, http.Header{}, []byte(teamLinkedDeactivatedBody))
|
||||
|
||||
require.True(t, shouldDisable)
|
||||
// fan-out 先标记同队兄弟（#2、#6），触发账户 #1 随后由常规 case 402 标记
|
||||
require.Equal(t, []int64{2, 6, 1}, repo.setErrorIDs)
|
||||
require.Contains(t, repo.setErrorMsgs[2], "team-linked error triggered by account #1")
|
||||
require.Contains(t, repo.setErrorMsgs[6], "team-linked error triggered by account #1")
|
||||
require.Contains(t, repo.setErrorMsgs[1], "Workspace deactivated (402)")
|
||||
require.NotContains(t, repo.setErrorMsgs[1], "team-linked")
|
||||
// 熔断顺序：兄弟账户先于落库全部进程内熔断，触发账户走 auth_error
|
||||
require.Equal(t, []string{openAITeamLinkedErrorBlockReason, openAITeamLinkedErrorBlockReason, "auth_error"}, blocker.reasons)
|
||||
require.Equal(t, int64(2), blocker.accounts[0].ID)
|
||||
require.Equal(t, int64(6), blocker.accounts[1].ID)
|
||||
require.Equal(t, int64(1), blocker.accounts[2].ID)
|
||||
}
|
||||
|
||||
func TestTeamLinkedError_GenericPaymentErrorDoesNotFanout(t *testing.T) {
|
||||
repo := &teamLinkedAccountRepoStub{teamAccounts: newTeamLinkedFixture()}
|
||||
rl, _ := newTeamLinkedTestService(repo)
|
||||
trigger := newTeamLinkedAccount(1, "team-A")
|
||||
|
||||
rl.HandleUpstreamError(context.Background(), &trigger, http.StatusPaymentRequired, http.Header{}, []byte(`{"error":{"message":"insufficient balance"}}`))
|
||||
|
||||
require.Equal(t, []int64{1}, repo.setErrorIDs)
|
||||
require.Contains(t, repo.setErrorMsgs[1], "Payment required (402)")
|
||||
require.Zero(t, repo.listCalls)
|
||||
}
|
||||
|
||||
func TestTeamLinkedError_DedupWithinTTL(t *testing.T) {
|
||||
repo := &teamLinkedAccountRepoStub{teamAccounts: newTeamLinkedFixture()}
|
||||
rl, _ := newTeamLinkedTestService(repo)
|
||||
first := newTeamLinkedAccount(1, "team-A")
|
||||
second := newTeamLinkedAccount(2, "team-A")
|
||||
|
||||
rl.HandleUpstreamError(context.Background(), &first, http.StatusPaymentRequired, http.Header{}, []byte(teamLinkedDeactivatedBody))
|
||||
rl.HandleUpstreamError(context.Background(), &second, http.StatusPaymentRequired, http.Header{}, []byte(teamLinkedDeactivatedBody))
|
||||
|
||||
// 第二次触发被去重：只有 #2 自身经 case 402 标记，未再次 fan-out
|
||||
require.Equal(t, []int64{2, 6, 1, 2}, repo.setErrorIDs)
|
||||
require.Equal(t, 1, repo.listCalls)
|
||||
}
|
||||
|
||||
func TestTeamLinkedError_APIKeyTriggerDoesNotFanout(t *testing.T) {
|
||||
repo := &teamLinkedAccountRepoStub{teamAccounts: newTeamLinkedFixture()}
|
||||
rl, _ := newTeamLinkedTestService(repo)
|
||||
trigger := newTeamLinkedAccount(4, "team-A")
|
||||
trigger.Type = AccountTypeAPIKey
|
||||
|
||||
rl.HandleUpstreamError(context.Background(), &trigger, http.StatusPaymentRequired, http.Header{}, []byte(teamLinkedDeactivatedBody))
|
||||
|
||||
require.Equal(t, []int64{4}, repo.setErrorIDs)
|
||||
require.Zero(t, repo.listCalls)
|
||||
}
|
||||
|
||||
func TestTeamLinkedError_DirectCallSkipsTriggerAccount(t *testing.T) {
|
||||
// 直调对应 fastpath 调用点：账户级临时不可调度规则短路时联动仍然生效
|
||||
repo := &teamLinkedAccountRepoStub{teamAccounts: newTeamLinkedFixture()}
|
||||
rl, blocker := newTeamLinkedTestService(repo)
|
||||
trigger := newTeamLinkedAccount(1, "team-A")
|
||||
|
||||
rl.maybeHandleOpenAITeamLinkedError(context.Background(), &trigger, http.StatusPaymentRequired, []byte(teamLinkedDeactivatedBody))
|
||||
|
||||
require.Equal(t, []int64{2, 6}, repo.setErrorIDs)
|
||||
require.Equal(t, []string{openAITeamLinkedErrorBlockReason, openAITeamLinkedErrorBlockReason}, blocker.reasons)
|
||||
}
|
||||
|
||||
func TestTeamLinkedError_MissingTeamIDDoesNothing(t *testing.T) {
|
||||
repo := &teamLinkedAccountRepoStub{teamAccounts: newTeamLinkedFixture()}
|
||||
rl, blocker := newTeamLinkedTestService(repo)
|
||||
trigger := Account{ID: 9, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Status: StatusActive}
|
||||
|
||||
rl.maybeHandleOpenAITeamLinkedError(context.Background(), &trigger, http.StatusPaymentRequired, []byte(teamLinkedDeactivatedBody))
|
||||
|
||||
require.Empty(t, repo.setErrorIDs)
|
||||
require.Empty(t, blocker.reasons)
|
||||
require.Zero(t, repo.listCalls)
|
||||
}
|
||||
|
||||
func TestTeamLinkedError_SetErrorFailureDoesNotAbortRemaining(t *testing.T) {
|
||||
repo := &teamLinkedAccountRepoStub{
|
||||
teamAccounts: newTeamLinkedFixture(),
|
||||
failSetError: map[int64]error{2: errors.New("db down")},
|
||||
}
|
||||
rl, blocker := newTeamLinkedTestService(repo)
|
||||
trigger := newTeamLinkedAccount(1, "team-A")
|
||||
|
||||
rl.maybeHandleOpenAITeamLinkedError(context.Background(), &trigger, http.StatusPaymentRequired, []byte(teamLinkedDeactivatedBody))
|
||||
|
||||
require.Equal(t, []int64{6}, repo.setErrorIDs)
|
||||
// 进程内熔断先于落库执行，两个账户都已被熔断
|
||||
require.Len(t, blocker.reasons, 2)
|
||||
}
|
||||
Reference in New Issue
Block a user