diff --git a/backend/internal/service/openai_account_runtime_block_fastpath.go b/backend/internal/service/openai_account_runtime_block_fastpath.go index ace8366a0..282237425 100644 --- a/backend/internal/service/openai_account_runtime_block_fastpath.go +++ b/backend/internal/service/openai_account_runtime_block_fastpath.go @@ -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 diff --git a/backend/internal/service/openai_team_linked_error.go b/backend/internal/service/openai_team_linked_error.go new file mode 100644 index 000000000..2225e59a5 --- /dev/null +++ b/backend/internal/service/openai_team_linked_error.go @@ -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 +} diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index ee5b09bbb..ec0c6bf19 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -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() // 池模式默认不标记本地账号状态;但管理员显式配置的临时不可调度规则优先。 diff --git a/backend/internal/service/ratelimit_service_openai_team_linked_test.go b/backend/internal/service/ratelimit_service_openai_team_linked_test.go new file mode 100644 index 000000000..e72763e26 --- /dev/null +++ b/backend/internal/service/ratelimit_service_openai_team_linked_test.go @@ -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) +}