diff --git a/backend/cmd/profit-preview/main.go b/backend/cmd/profit-preview/main.go index 6cb53d331f..2c351410c9 100644 --- a/backend/cmd/profit-preview/main.go +++ b/backend/cmd/profit-preview/main.go @@ -134,6 +134,9 @@ func main() { for _, model := range modelsWithZeroRemaining(report) { fmt.Printf(" 警告: 模型 %s 启用后利润门准入账号为 0\n", model) } + for _, model := range modelsWithZeroRemainingUnderMinD(report) { + fmt.Printf(" 警告: 模型 %s 在最低有效D（存在低倍率用户覆盖）下利润门准入账号为 0\n", model) + } } } @@ -217,3 +220,18 @@ func modelsWithZeroRemaining(report service.ProfitPreviewGroupReport) []string { sort.Strings(out) return out } + +// modelsWithZeroRemainingUnderMinD 返回默认 D 下仍有准入账号、但在最低有效 D +// 下会归零的模型。最低有效 D 来自分组内最低的用户级倍率覆盖：这些模型对那部分 +// 用户是全黑的，而只看默认 D 的告警完全看不出来。 +// 两档都为 0 的模型由 modelsWithZeroRemaining 报告，这里不重复。 +func modelsWithZeroRemainingUnderMinD(report service.ProfitPreviewGroupReport) []string { + var out []string + for model, count := range report.RemainingByModelMinD { + if count == 0 && report.RemainingByModel[model] > 0 { + out = append(out, model) + } + } + sort.Strings(out) + return out +} diff --git a/backend/cmd/profit-preview/main_test.go b/backend/cmd/profit-preview/main_test.go index 5607ecb09e..9b6d185c5b 100644 --- a/backend/cmd/profit-preview/main_test.go +++ b/backend/cmd/profit-preview/main_test.go @@ -54,3 +54,39 @@ func TestParsePreviewInputsRejectsEmptyGroups(t *testing.T) { require.Nil(t, inputs) } } + +// TestModelsWithZeroRemainingWarnings 钉死两档 D 的归零告警分工： +// 默认 D 归零由 modelsWithZeroRemaining 报告；默认 D 仍有账号但最低有效 D +// （分组内存在更低的用户级倍率覆盖）归零的模型必须单独告警——那些用户的该 +// 模型会全黑，只看默认 D 完全看不出来。两档都为 0 时不重复告警。 +func TestModelsWithZeroRemainingWarnings(t *testing.T) { + report := service.ProfitPreviewGroupReport{ + RemainingByModel: map[string]int{ + "both-zero": 0, + "min-d-zero": 2, + "healthy": 3, + "min-d-zero-alt": 1, + }, + RemainingByModelMinD: map[string]int{ + "both-zero": 0, + "min-d-zero": 0, + "healthy": 3, + "min-d-zero-alt": 0, + }, + } + + if got := modelsWithZeroRemaining(report); len(got) != 1 || got[0] != "both-zero" { + t.Fatalf("默认D归零告警应只覆盖 both-zero，got %v", got) + } + + got := modelsWithZeroRemainingUnderMinD(report) + want := []string{"min-d-zero", "min-d-zero-alt"} + if len(got) != len(want) { + t.Fatalf("最低有效D归零告警不符: got %v want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("最低有效D归零告警不符（应按模型名排序）: got %v want %v", got, want) + } + } +} diff --git a/backend/ent/schema/group.go b/backend/ent/schema/group.go index 49c0f63151..8a9fc12ded 100644 --- a/backend/ent/schema/group.go +++ b/backend/ent/schema/group.go @@ -235,7 +235,7 @@ func (Group) Fields() []ent.Field { SchemaType(map[string]string{dialect.Postgres: "jsonb"}). Comment("OpenAI reasoning effort 自定义精确映射；先映射再应用上限"), - // 分组利润控制（migration 191）：openai/anthropic/gemini/grok/antigravity + // 分组利润控制（migration 192/193）：openai/anthropic/gemini/grok/antigravity // 的 token 分组可启用，composite 分组不能直接启用。 field.Bool("profit_control_enabled"). Default(false). diff --git a/backend/internal/handler/admin/group_handler.go b/backend/internal/handler/admin/group_handler.go index 11662cbbef..e99ba7b755 100644 --- a/backend/internal/handler/admin/group_handler.go +++ b/backend/internal/handler/admin/group_handler.go @@ -481,7 +481,9 @@ func (h *GroupHandler) Create(c *gin.Context) { return } - if err := service.ValidateProfitControlConfig(req.Platform, req.ProfitControlEnabled, float64ValueOrDefault(req.ProfitMinMargin, 0), float64ValueOrDefault(req.ProfitSafetyBuffer, 0)); err != nil { + // platform 是 omitempty：预校验必须用与 CreateGroup 落库一致的归一化平台， + // 否则省略 platform 的请求会被误判成「平台不支持利润控制」。 + if err := service.ValidateProfitControlConfig(service.NormalizeGroupPlatform(req.Platform), req.ProfitControlEnabled, float64ValueOrDefault(req.ProfitMinMargin, 0), float64ValueOrDefault(req.ProfitSafetyBuffer, 0)); err != nil { response.BadRequest(c, err.Error()) return } diff --git a/backend/internal/handler/dto/group_mapper_profit_visibility_test.go b/backend/internal/handler/dto/group_mapper_profit_visibility_test.go new file mode 100644 index 0000000000..a3a9c5a83f --- /dev/null +++ b/backend/internal/handler/dto/group_mapper_profit_visibility_test.go @@ -0,0 +1,75 @@ +package dto + +import ( + "encoding/json" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/service" +) + +// profitControlJSONFields 是分组利润控制的三个 JSON 字段。它们与同响应中的 +// rate_multiplier 相乘即可反推出运营方的上游采购成本上限，属于内部经营信息， +// 只能出现在管理员 DTO 中。 +var profitControlJSONFields = []string{ + "profit_control_enabled", + "profit_min_margin", + "profit_safety_buffer", +} + +func profitControlServiceGroup() *service.Group { + return &service.Group{ + ID: 7, + Name: "profit-gated", + Platform: service.PlatformAnthropic, + RateMultiplier: 2.0, + Status: service.StatusActive, + ProfitControlEnabled: true, + ProfitMinMargin: 0.3, + ProfitSafetyBuffer: 0.05, + } +} + +func marshalToMap(t *testing.T, v any) map[string]any { + t.Helper() + raw, err := json.Marshal(v) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var out map[string]any + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatalf("unmarshal: %v", err) + } + return out +} + +// TestGroupFromServiceOmitsProfitControl 钉死普通用户侧的分组 DTO 不泄露利润控制配置。 +func TestGroupFromServiceOmitsProfitControl(t *testing.T) { + for name, got := range map[string]any{ + "GroupFromService": GroupFromService(profitControlServiceGroup()), + "GroupFromServiceShallow": GroupFromServiceShallow(profitControlServiceGroup()), + } { + fields := marshalToMap(t, got) + for _, f := range profitControlJSONFields { + if _, ok := fields[f]; ok { + t.Errorf("%s: 普通用户 DTO 不得包含 %q", name, f) + } + } + if _, ok := fields["rate_multiplier"]; !ok { + t.Errorf("%s: 应仍返回 rate_multiplier", name) + } + } +} + +// TestGroupFromServiceAdminIncludesProfitControl 钉死管理端仍能读写利润控制配置。 +func TestGroupFromServiceAdminIncludesProfitControl(t *testing.T) { + admin := GroupFromServiceAdmin(profitControlServiceGroup()) + if admin.ProfitControlEnabled != true || admin.ProfitMinMargin != 0.3 || admin.ProfitSafetyBuffer != 0.05 { + t.Fatalf("管理员 DTO 未透传利润控制配置: %+v", admin) + } + fields := marshalToMap(t, admin) + for _, f := range profitControlJSONFields { + if _, ok := fields[f]; !ok { + t.Errorf("管理员 DTO 应包含 %q", f) + } + } +} diff --git a/backend/internal/handler/dto/mappers.go b/backend/internal/handler/dto/mappers.go index 980f92deb9..c6f6cf25e2 100644 --- a/backend/internal/handler/dto/mappers.go +++ b/backend/internal/handler/dto/mappers.go @@ -146,6 +146,9 @@ func GroupFromServiceAdmin(g *service.Group) *AdminGroup { } out := &AdminGroup{ Group: groupFromServiceBase(g), + ProfitControlEnabled: g.ProfitControlEnabled, + ProfitMinMargin: g.ProfitMinMargin, + ProfitSafetyBuffer: g.ProfitSafetyBuffer, ModelRouting: g.ModelRouting, ModelRoutingEnabled: g.ModelRoutingEnabled, MCPXMLInject: g.MCPXMLInject, @@ -193,9 +196,6 @@ func groupFromServiceBase(g *service.Group) Group { PeakStart: g.PeakStart, PeakEnd: g.PeakEnd, PeakRateMultiplier: g.PeakRateMultiplier, - ProfitControlEnabled: g.ProfitControlEnabled, - ProfitMinMargin: g.ProfitMinMargin, - ProfitSafetyBuffer: g.ProfitSafetyBuffer, ImagePrice1K: g.ImagePrice1K, ImagePrice2K: g.ImagePrice2K, ImagePrice4K: g.ImagePrice4K, diff --git a/backend/internal/handler/dto/types.go b/backend/internal/handler/dto/types.go index b990b57894..37927c2b48 100644 --- a/backend/internal/handler/dto/types.go +++ b/backend/internal/handler/dto/types.go @@ -111,20 +111,16 @@ type Group struct { VideoRateIndependent bool `json:"video_rate_independent"` VideoRateMultiplier float64 `json:"video_rate_multiplier"` // 高峰时段倍率配置 - PeakRateEnabled bool `json:"peak_rate_enabled"` - PeakStart string `json:"peak_start"` - PeakEnd string `json:"peak_end"` - PeakRateMultiplier float64 `json:"peak_rate_multiplier"` - // 分组利润控制（五个 token 平台分组可启用；margin/buffer 为小数存储） - ProfitControlEnabled bool `json:"profit_control_enabled"` - ProfitMinMargin float64 `json:"profit_min_margin"` - ProfitSafetyBuffer float64 `json:"profit_safety_buffer"` - ImagePrice1K *float64 `json:"image_price_1k"` - ImagePrice2K *float64 `json:"image_price_2k"` - ImagePrice4K *float64 `json:"image_price_4k"` - VideoPrice480P *float64 `json:"video_price_480p"` - VideoPrice720P *float64 `json:"video_price_720p"` - VideoPrice1080P *float64 `json:"video_price_1080p"` + PeakRateEnabled bool `json:"peak_rate_enabled"` + PeakStart string `json:"peak_start"` + PeakEnd string `json:"peak_end"` + PeakRateMultiplier float64 `json:"peak_rate_multiplier"` + ImagePrice1K *float64 `json:"image_price_1k"` + ImagePrice2K *float64 `json:"image_price_2k"` + ImagePrice4K *float64 `json:"image_price_4k"` + VideoPrice480P *float64 `json:"video_price_480p"` + VideoPrice720P *float64 `json:"video_price_720p"` + VideoPrice1080P *float64 `json:"video_price_1080p"` // Codex alpha/search 网页搜索单次价格（USD/次）；null 表示使用默认价 0.01 WebSearchPricePerCall *float64 `json:"web_search_price_per_call"` @@ -159,6 +155,13 @@ type Group struct { type AdminGroup struct { Group + // 分组利润控制（五个 token 平台分组可启用；margin/buffer 为小数存储）。 + // 仅管理员可见：这三个字段与同响应中的 rate_multiplier 相乘即可反推出 + // 运营方的上游成本上限，属于内部经营信息，不得下放到 dto.Group。 + ProfitControlEnabled bool `json:"profit_control_enabled"` + ProfitMinMargin float64 `json:"profit_min_margin"` + ProfitSafetyBuffer float64 `json:"profit_safety_buffer"` + // 模型路由配置（仅 anthropic 平台使用） ModelRouting map[string][]int64 `json:"model_routing"` ModelRoutingEnabled bool `json:"model_routing_enabled"` diff --git a/backend/internal/handler/failover_loop.go b/backend/internal/handler/failover_loop.go index 6b4858c528..951a2f6b35 100644 --- a/backend/internal/handler/failover_loop.go +++ b/backend/internal/handler/failover_loop.go @@ -40,8 +40,18 @@ const ( // Service 层在 SingleAccountRetry 模式下已做充分原地重试（最多 3 次、总等待 30s）， // Handler 层只需短暂间隔后重新进入 Service 层即可。 singleAccountBackoffDelay = 2 * time.Second + // maxProfitVetoAttempts 单次请求内允许的分组利润门终检否决次数上限。 + // 利润否决不产生上游请求，因此不会推进 SwitchCount；没有独立上限的话， + // 「选号 → 终检否决 → 重选」在候选池与账号快照短暂不一致时可以空转很久。 + // 取值与 maxAccountSwitches 默认值一致：混合定价的大分组仍有充分重选机会， + // 同时把整池越线时的无谓选号开销限制在常数级。 + maxProfitVetoAttempts = 10 ) +// profitVetoExhaustedMessage 是利润否决次数耗尽时返回给客户端的文案。 +// 语义上等同于「无可用账号」：候选账号都不满足分组的利润约束。 +const profitVetoExhaustedMessage = "No available accounts: all candidates rejected by group profit control" + // FailoverState 跨循环迭代共享的 failover 状态 type FailoverState struct { SwitchCount int @@ -51,18 +61,64 @@ type FailoverState struct { LastFailoverErr *service.UpstreamFailoverError ForceCacheBilling bool hasBoundSession bool + + // profitVetoedAccountIDs 记录被分组利润门终检否决的账号，是 FailedAccountIDs + // 的子集。之所以单独维护：HandleSelectionExhausted 的 503 退避分支会清空 + // FailedAccountIDs，而利润否决在同一请求内的判定不会改变（下游倍率 D 已在 + // 请求开始冻结），被清空的账号会被立即重选并再次否决，形成没有任何上游请求、 + // SwitchCount 也不前进的活锁。清空后必须把它们放回排除集。 + profitVetoedAccountIDs map[int64]struct{} + // profitVetoCount 本次请求累计的利润否决次数，用于 maxProfitVetoAttempts 上限。 + profitVetoCount int } // NewFailoverState 创建 failover 状态 func NewFailoverState(maxSwitches int, hasBoundSession bool) *FailoverState { return &FailoverState{ - MaxSwitches: maxSwitches, - FailedAccountIDs: make(map[int64]struct{}), - SameAccountRetryCount: make(map[int64]int), - hasBoundSession: hasBoundSession, + MaxSwitches: maxSwitches, + FailedAccountIDs: make(map[int64]struct{}), + SameAccountRetryCount: make(map[int64]int), + hasBoundSession: hasBoundSession, + profitVetoedAccountIDs: make(map[int64]struct{}), } } +// RecordProfitVeto 记录一次分组利润门终检否决：把账号加入排除列表（同时登记到 +// 利润否决集，使其不被 503 退避分支清掉）并递增否决计数。 +// +// 返回 FailoverContinue 表示调用方可以继续重选下一个账号；返回 FailoverExhausted +// 表示本次请求的利润否决次数已达上限，调用方应按「无可用账号」终止， +// 不得继续 continue。 +func (s *FailoverState) RecordProfitVeto(accountID int64) FailoverAction { + s.FailedAccountIDs[accountID] = struct{}{} + if s.profitVetoedAccountIDs == nil { + s.profitVetoedAccountIDs = make(map[int64]struct{}) + } + s.profitVetoedAccountIDs[accountID] = struct{}{} + s.profitVetoCount++ + if s.profitVetoCount >= maxProfitVetoAttempts { + return FailoverExhausted + } + return FailoverContinue +} + +// ProfitVetoCount 返回本次请求累计的利润否决次数（供日志使用）。 +func (s *FailoverState) ProfitVetoCount() int { return s.profitVetoCount } + +// allExclusionsAreProfitVetoed 判断排除列表是否已全部由利润门否决贡献。 +// 此时清空 FailedAccountIDs 会被原样恢复，退避重试不会带来任何新候选。 +func (s *FailoverState) allExclusionsAreProfitVetoed() bool { + if len(s.profitVetoedAccountIDs) == 0 || len(s.FailedAccountIDs) == 0 { + return false + } + for id := range s.FailedAccountIDs { + if _, ok := s.profitVetoedAccountIDs[id]; !ok { + return false + } + } + return true +} + // HandleFailoverError 处理 UpstreamFailoverError，返回下一步动作。 // 包含：缓存计费判断、同账号重试、临时封禁、切换计数、Antigravity 延时。 func (s *FailoverState) HandleFailoverError( @@ -156,6 +212,17 @@ func (s *FailoverState) HandleSelectionExhausted(ctx context.Context) FailoverAc s.LastFailoverErr.StatusCode == http.StatusServiceUnavailable && s.SwitchCount <= s.MaxSwitches { + // 排除列表全由利润门否决贡献时，清空后会被原样恢复：退避重试拿不到 + // 任何新候选，而利润否决不推进 SwitchCount，退避条件将永远成立。 + // 这里直接判定耗尽，避免每 2s 空转一轮的活锁。 + if s.allExclusionsAreProfitVetoed() { + logger.FromContext(ctx).Warn("gateway.failover_selection_exhausted_by_profit_veto", + zap.Int("profit_veto_count", s.profitVetoCount), + zap.Int("excluded_accounts", len(s.FailedAccountIDs)), + ) + return FailoverExhausted + } + logger.FromContext(ctx).Warn("gateway.failover_single_account_backoff", zap.Duration("backoff_delay", singleAccountBackoffDelay), zap.Int("switch_count", s.SwitchCount), @@ -169,6 +236,11 @@ func (s *FailoverState) HandleSelectionExhausted(ctx context.Context) FailoverAc zap.Int("max_switches", s.MaxSwitches), ) s.FailedAccountIDs = make(map[int64]struct{}) + // 利润门否决的账号不参与退避重试的解除：判定依据（冻结的下游倍率）在 + // 同一请求内不变，放它们回池只会被再次否决。 + for id := range s.profitVetoedAccountIDs { + s.FailedAccountIDs[id] = struct{}{} + } return FailoverContinue } return FailoverExhausted diff --git a/backend/internal/handler/failover_loop_profit_veto_test.go b/backend/internal/handler/failover_loop_profit_veto_test.go new file mode 100644 index 0000000000..c417c27e10 --- /dev/null +++ b/backend/internal/handler/failover_loop_profit_veto_test.go @@ -0,0 +1,139 @@ +package handler + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// profitVetoLoopResult 记录一次模拟选号循环的终止方式与步数。 +type profitVetoLoopResult struct { + outcome string // "forwarded" | "exhausted" | "budget_exceeded" + forwardedID int64 + iterations int + backoffRetries int +} + +// runProfitVetoLoop 模拟 handler 的「选号 → 利润终检 → 重选」循环，只保留与 +// 活锁相关的状态机（FailoverState + 排除列表 + HandleSelectionExhausted）， +// 不涉及真实调度器与上游转发。 +// +// pool 按顺序给出候选账号；vetoed 中的账号每次终检都被利润门否决——这对应 +// 「候选池快照与 per-account 快照短暂不一致」时门的确定性判定。 +// maxIterations 是测试自身的预算：循环超过它即视为活锁。 +func runProfitVetoLoop(t *testing.T, fs *FailoverState, pool []int64, vetoed map[int64]bool, maxIterations int) profitVetoLoopResult { + t.Helper() + res := profitVetoLoopResult{} + for res.iterations = 1; res.iterations <= maxIterations; res.iterations++ { + // 选号：返回第一个不在排除列表中的账号。 + var picked int64 + for _, id := range pool { + if _, excluded := fs.FailedAccountIDs[id]; !excluded { + picked = id + break + } + } + if picked == 0 { + // 选号耗尽，交给退避决策。 + switch fs.HandleSelectionExhausted(context.Background()) { + case FailoverContinue: + res.backoffRetries++ + continue + default: + res.outcome = "exhausted" + return res + } + } + if vetoed[picked] { + if fs.RecordProfitVeto(picked) == FailoverExhausted { + res.outcome = "exhausted" + return res + } + continue + } + res.outcome = "forwarded" + res.forwardedID = picked + return res + } + res.outcome = "budget_exceeded" + return res +} + +// TestProfitVetoAfter503DoesNotLivelock 钉死 #4925 引入的活锁回归： +// 一次真实 503 之后，若调度器持续返回会被利润门否决的账号， +// 「选号 → 否决 → 排除 → 选号耗尽 → 清空排除 → 睡 2s → 选号」会无限循环， +// 因为利润否决不推进 SwitchCount，退避条件永远成立。 +func TestProfitVetoAfter503DoesNotLivelock(t *testing.T) { + fs := NewFailoverState(10, false) + // 已经历一次真实 503（Antigravity 单账号分组 MODEL_CAPACITY_EXHAUSTED 是设计内路径）。 + fs.LastFailoverErr = newTestFailoverErr(503, false, false) + fs.SwitchCount = 1 + fs.FailedAccountIDs[1] = struct{}{} + + start := time.Now() + res := runProfitVetoLoop(t, fs, []int64{1}, map[int64]bool{1: true}, 50) + elapsed := time.Since(start) + + require.Equal(t, "exhausted", res.outcome, "整池被利润门否决时必须有限步终止") + require.LessOrEqual(t, res.backoffRetries, 1, "被否决的账号不得被退避分支复活并反复重选") + require.Less(t, elapsed, 10*time.Second, "不得每 2s 空转一轮") +} + +// TestProfitVetoKeepsBackoffUsefulForHealthyAccount 钉死修复没有削弱既有的 +// 503 退避语义：排除列表里仍有非利润否决的账号时，退避照常清空并重试， +// 只是被利润门否决的账号不再复活。 +func TestProfitVetoKeepsBackoffUsefulForHealthyAccount(t *testing.T) { + fs := NewFailoverState(10, false) + fs.LastFailoverErr = newTestFailoverErr(503, false, false) + fs.SwitchCount = 1 + // 账号 1 因真实 503 被排除；账号 2 会被利润门否决。 + fs.FailedAccountIDs[1] = struct{}{} + + res := runProfitVetoLoop(t, fs, []int64{2, 1}, map[int64]bool{2: true}, 50) + + require.Equal(t, "forwarded", res.outcome) + require.Equal(t, int64(1), res.forwardedID, "退避清空后应重新可选账号 1") + require.Equal(t, 1, res.backoffRetries, "应发生且只发生一次退避重试") + require.Contains(t, fs.FailedAccountIDs, int64(2), "利润否决的账号在退避清空后必须被放回排除集") +} + +// TestProfitVetoAttemptsCapped 钉死没有 503 参与时，大分组整池越线也会在 +// 常数步内终止，而不是把整池逐个选一遍。 +func TestProfitVetoAttemptsCapped(t *testing.T) { + fs := NewFailoverState(10, false) + pool := make([]int64, 0, 64) + vetoed := make(map[int64]bool, 64) + for id := int64(1); id <= 64; id++ { + pool = append(pool, id) + vetoed[id] = true + } + + res := runProfitVetoLoop(t, fs, pool, vetoed, 200) + + require.Equal(t, "exhausted", res.outcome) + require.Equal(t, maxProfitVetoAttempts, fs.ProfitVetoCount()) + require.Equal(t, maxProfitVetoAttempts, res.iterations, "达到上限即终止，不应继续遍历候选池") +} + +// TestRecordProfitVetoExcludesAccount 钉死 RecordProfitVeto 仍然把账号加入 +// 调度排除列表（选号入参用的就是 FailedAccountIDs）。 +func TestRecordProfitVetoExcludesAccount(t *testing.T) { + fs := NewFailoverState(10, false) + require.Equal(t, FailoverContinue, fs.RecordProfitVeto(42)) + require.Contains(t, fs.FailedAccountIDs, int64(42)) + require.Equal(t, 1, fs.ProfitVetoCount()) +} + +// TestHandleSelectionExhaustedUnaffectedWithoutProfitVeto 钉死未启用利润控制的 +// 请求（无任何利润否决）走的仍是原有退避语义：清空排除列表并重试。 +func TestHandleSelectionExhaustedUnaffectedWithoutProfitVeto(t *testing.T) { + fs := NewFailoverState(3, false) + fs.LastFailoverErr = newTestFailoverErr(503, false, false) + fs.SwitchCount = 1 + fs.FailedAccountIDs[100] = struct{}{} + + require.Equal(t, FailoverContinue, fs.HandleSelectionExhausted(context.Background())) + require.Empty(t, fs.FailedAccountIDs, "无利润否决时排除列表应被完全清空") +} diff --git a/backend/internal/handler/gateway_handler.go b/backend/internal/handler/gateway_handler.go index 3d9d1e4d03..287d59fd9f 100644 --- a/backend/internal/handler/gateway_handler.go +++ b/backend/internal/handler/gateway_handler.go @@ -432,7 +432,12 @@ func (h *GatewayHandler) Messages(c *gin.Context) { accountReleaseFunc() } reqLog.Debug("gateway.account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) - fs.FailedAccountIDs[account.ID] = struct{}{} + if fs.RecordProfitVeto(account.ID) == FailoverExhausted { + reqLog.Warn("gateway.profit_veto_attempts_exhausted", zap.Int("profit_veto_count", fs.ProfitVetoCount())) + markOpsRoutingCapacityLimited(c) + h.handleStreamingAwareError(c, http.StatusServiceUnavailable, "api_error", profitVetoExhaustedMessage, streamStarted) + return + } continue } account = latest @@ -750,7 +755,12 @@ func (h *GatewayHandler) Messages(c *gin.Context) { accountReleaseFunc() } reqLog.Debug("gateway.account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) - fs.FailedAccountIDs[account.ID] = struct{}{} + if fs.RecordProfitVeto(account.ID) == FailoverExhausted { + reqLog.Warn("gateway.profit_veto_attempts_exhausted", zap.Int("profit_veto_count", fs.ProfitVetoCount())) + markOpsRoutingCapacityLimited(c) + h.handleStreamingAwareError(c, http.StatusServiceUnavailable, "api_error", profitVetoExhaustedMessage, streamStarted) + return + } continue } account = latest diff --git a/backend/internal/handler/gateway_handler_chat_completions.go b/backend/internal/handler/gateway_handler_chat_completions.go index 294baae826..ea4e839636 100644 --- a/backend/internal/handler/gateway_handler_chat_completions.go +++ b/backend/internal/handler/gateway_handler_chat_completions.go @@ -232,7 +232,11 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) { accountReleaseFunc() } reqLog.Debug("gateway.cc.account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) - fs.FailedAccountIDs[account.ID] = struct{}{} + if fs.RecordProfitVeto(account.ID) == FailoverExhausted { + reqLog.Warn("gateway.cc.profit_veto_attempts_exhausted", zap.Int("profit_veto_count", fs.ProfitVetoCount())) + h.chatCompletionsErrorResponse(c, http.StatusServiceUnavailable, "api_error", profitVetoExhaustedMessage) + return + } continue } account = latest diff --git a/backend/internal/handler/gateway_handler_responses.go b/backend/internal/handler/gateway_handler_responses.go index 0102fa9f0d..7f49f39a3e 100644 --- a/backend/internal/handler/gateway_handler_responses.go +++ b/backend/internal/handler/gateway_handler_responses.go @@ -236,7 +236,11 @@ func (h *GatewayHandler) Responses(c *gin.Context) { accountReleaseFunc() } reqLog.Debug("gateway.responses.account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) - fs.FailedAccountIDs[account.ID] = struct{}{} + if fs.RecordProfitVeto(account.ID) == FailoverExhausted { + reqLog.Warn("gateway.responses.profit_veto_attempts_exhausted", zap.Int("profit_veto_count", fs.ProfitVetoCount())) + h.responsesErrorResponse(c, http.StatusServiceUnavailable, "api_error", profitVetoExhaustedMessage) + return + } continue } account = latest diff --git a/backend/internal/handler/gemini_v1beta_handler.go b/backend/internal/handler/gemini_v1beta_handler.go index 45136b306d..2a270d3a0a 100644 --- a/backend/internal/handler/gemini_v1beta_handler.go +++ b/backend/internal/handler/gemini_v1beta_handler.go @@ -477,7 +477,12 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) { accountReleaseFunc() } reqLog.Debug("gemini.account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) - fs.FailedAccountIDs[account.ID] = struct{}{} + if fs.RecordProfitVeto(account.ID) == FailoverExhausted { + reqLog.Warn("gemini.profit_veto_attempts_exhausted", zap.Int("profit_veto_count", fs.ProfitVetoCount())) + markOpsRoutingCapacityLimited(c) + googleError(c, http.StatusServiceUnavailable, profitVetoExhaustedMessage) + return + } continue } account = latest diff --git a/backend/internal/handler/grok_media.go b/backend/internal/handler/grok_media.go index 15a93de24a..e11bf7f08c 100644 --- a/backend/internal/handler/grok_media.go +++ b/backend/internal/handler/grok_media.go @@ -181,6 +181,7 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. // 范围内：显式豁免，防止 service 层防御性装门按文本 D 误过滤媒体请求， // 也防止已计费的在途视频任务因绑定账号被门排除而查询返回伪 404。 requestCtx := service.WithOpenAIProfitControlSuppressed(c.Request.Context()) + profitVetoCount := 0 failedAccountIDs := make(map[int64]struct{}) sameAccountRetryCount := make(map[int64]int) var lastFailoverErr *service.UpstreamFailoverError @@ -298,8 +299,12 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service. accountReleaseFunc, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, false, &streamStarted, reqLog) if slotResult == openAISlotAcquireProfitVetoed { - // 媒体路径已显式豁免利润门（suppress 标记），此分支仅防御性兜底。 - failedAccountIDs[account.ID] = struct{}{} + // 媒体路径已显式豁免利润门（suppress 标记），此分支仅防御性兜底， + // 同样受否决上限约束。 + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + h.handleOpenAIProfitVetoExhausted(c, streamStarted, reqLog, profitVetoCount) + return + } continue } if slotResult != openAISlotAcquireOK { diff --git a/backend/internal/handler/openai_alpha_search.go b/backend/internal/handler/openai_alpha_search.go index 4c11a01555..c5766e5669 100644 --- a/backend/internal/handler/openai_alpha_search.go +++ b/backend/internal/handler/openai_alpha_search.go @@ -107,6 +107,7 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) { searchID := strings.TrimSpace(gjson.GetBytes(body, "id").String()) sessionHash := h.gatewayService.GenerateSessionHashWithFallback(c, nil, searchID) + profitVetoCount := 0 failedAccountIDs := make(map[int64]struct{}) var lastFailoverErr *service.UpstreamFailoverError switchCount := 0 @@ -158,8 +159,11 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) { setOpsSelectedAccount(c, account.ID, account.Platform) accountRelease, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, false, &streamStarted, reqLog) if slotResult == openAISlotAcquireProfitVetoed { - // 利润终检否决：排除该账号重新选号，全池耗尽由下一轮选号报错。 - failedAccountIDs[account.ID] = struct{}{} + // 利润终检否决：排除该账号重新选号；否决次数达上限则按无可用账号终止。 + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + h.handleOpenAIProfitVetoExhausted(c, streamStarted, reqLog, profitVetoCount) + return + } continue } if slotResult != openAISlotAcquireOK { diff --git a/backend/internal/handler/openai_chat_completions.go b/backend/internal/handler/openai_chat_completions.go index 0d03e034f3..efd67088d1 100644 --- a/backend/internal/handler/openai_chat_completions.go +++ b/backend/internal/handler/openai_chat_completions.go @@ -144,6 +144,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { maxAccountSwitches := h.maxAccountSwitches switchCount := 0 + profitVetoCount := 0 failedAccountIDs := make(map[int64]struct{}) sameAccountRetryCount := make(map[int64]int) var lastFailoverErr *service.UpstreamFailoverError @@ -213,8 +214,11 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) { accountReleaseFunc, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, reqStream, &streamStarted, reqLog) if slotResult == openAISlotAcquireProfitVetoed { - // 利润终检否决：排除该账号重新选号，全池耗尽由下一轮选号报错。 - failedAccountIDs[account.ID] = struct{}{} + // 利润终检否决：排除该账号重新选号；否决次数达上限则按无可用账号终止。 + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + h.handleOpenAIProfitVetoExhausted(c, streamStarted, reqLog, profitVetoCount) + return + } continue } if slotResult != openAISlotAcquireOK { diff --git a/backend/internal/handler/openai_embeddings.go b/backend/internal/handler/openai_embeddings.go index 531f7e137f..92008c59c2 100644 --- a/backend/internal/handler/openai_embeddings.go +++ b/backend/internal/handler/openai_embeddings.go @@ -107,6 +107,7 @@ func (h *OpenAIGatewayHandler) Embeddings(c *gin.Context) { return } + profitVetoCount := 0 failedAccountIDs := make(map[int64]struct{}) var lastFailoverErr *service.UpstreamFailoverError switchCount := 0 @@ -171,8 +172,11 @@ func (h *OpenAIGatewayHandler) Embeddings(c *gin.Context) { accountReleaseFunc, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, "", selection, false, &streamStarted, reqLog) if slotResult == openAISlotAcquireProfitVetoed { - // 利润终检否决：排除该账号重新选号，全池耗尽由下一轮选号报错。 - failedAccountIDs[account.ID] = struct{}{} + // 利润终检否决：排除该账号重新选号；否决次数达上限则按无可用账号终止。 + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + h.handleOpenAIProfitVetoExhausted(c, streamStarted, reqLog, profitVetoCount) + return + } continue } if slotResult != openAISlotAcquireOK { diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index 17ffb1b28e..74038fb284 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -427,6 +427,7 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { maxAccountSwitches := h.maxAccountSwitches switchCount := 0 firstOutputTimeoutSwitchCount := 0 + profitVetoCount := 0 failedAccountIDs := make(map[int64]struct{}) sameAccountRetryCount := make(map[int64]int) var lastFailoverErr *service.UpstreamFailoverError @@ -526,8 +527,12 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) { accountReleaseFunc, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, reqStream, &streamStarted, reqLog) if slotResult == openAISlotAcquireProfitVetoed { - // 利润终检否决：排除该账号重新选号，全池耗尽由下一轮选号报错。 - failedAccountIDs[account.ID] = struct{}{} + // 利润终检否决：排除该账号重新选号，全池耗尽由下一轮选号报错； + // 否决次数达上限则直接终止，避免排队抢槽后才终检的延迟放大。 + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + h.handleOpenAIProfitVetoExhausted(c, streamStarted, reqLog, profitVetoCount) + return + } continue } if slotResult != openAISlotAcquireOK { @@ -1002,6 +1007,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { maxAccountSwitches := h.maxAccountSwitches switchCount := 0 + profitVetoCount := 0 failedAccountIDs := make(map[int64]struct{}) sameAccountRetryCount := make(map[int64]int) var lastFailoverErr *service.UpstreamFailoverError @@ -1078,8 +1084,12 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) { accountReleaseFunc, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, reqStream, &streamStarted, reqLog) if slotResult == openAISlotAcquireProfitVetoed { - // 利润终检否决：排除该账号重新选号，全池耗尽由下一轮选号报错。 - failedAccountIDs[account.ID] = struct{}{} + // 利润终检否决：排除该账号重新选号，全池耗尽由下一轮选号报错； + // 否决次数达上限则直接终止，避免排队抢槽后才终检的延迟放大。 + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + h.handleOpenAIProfitVetoExhausted(c, streamStarted, reqLog, profitVetoCount) + return + } continue } if slotResult != openAISlotAcquireOK { @@ -1380,11 +1390,66 @@ const ( // openAISlotAcquireFailed：错误响应已写出，调用方直接 return。 openAISlotAcquireFailed // openAISlotAcquireProfitVetoed：槽位获取成功后利润终检否决。槽位已释放、 - // 未写任何响应；调用方应把该账号加入本请求排除集并重新选号，全池耗尽由 - // 下一轮选号返回标准 no available accounts。 + // 未写任何响应；调用方应经 recordOpenAIProfitVeto 把该账号加入本请求排除集 + // 后重新选号，全池耗尽由下一轮选号返回标准 no available accounts。 openAISlotAcquireProfitVetoed ) +// openAIWSTurnPricing 持有 WebSocket 连接内「当前 turn」的计费定价时刻。 +// 由 BeforeTurn 在每个 turn 开始时冻结，AfterTurn 的用量提交读取它；turn 在 +// 连接内串行推进，互斥锁只为跨用量提交 goroutine 的读取安全。 +// +// 零值语义（重要）：ws_v2 passthrough ingress 只实现了 AfterTurn，没有任何 +// turn 起始回调，BeforeTurn 永远不会被调用。此时本值保持零，RecordUsage 经 +// openAIUsagePricingAt 回退到记录时刻——与引入分组利润控制前的基线一致。 +// 绝不能用建连时刻初始化：那会把透传连接的所有 turn 钉死在建连时的高峰因子， +// 客户端只要峰前一分钟建连并保活，整条连接就能全程按谷价结算，正是利润控制 +// 想堵的漏洞。透传 ingress 目前不做 turn 级利润复核，只有建连时的准入门。 +type openAIWSTurnPricing struct { + mu sync.Mutex + at time.Time +} + +func (p *openAIWSTurnPricing) freeze(at time.Time) { + p.mu.Lock() + p.at = at + p.mu.Unlock() +} + +func (p *openAIWSTurnPricing) current() time.Time { + p.mu.Lock() + defer p.mu.Unlock() + return p.at +} + +// recordOpenAIProfitVeto 记录 OpenAI 侧选号循环的一次利润门终检否决：把账号 +// 加入本请求排除集并递增否决计数。返回 false 表示否决次数已达 +// maxProfitVetoAttempts，调用方必须停止重选并按「无可用账号」终止。 +// +// OpenAI 路径用的是各自的 failedAccountIDs map + for 循环（不是 FailoverState）， +// 这里用一个独立计数器复用同一上限语义。上限是必需的：WaitPlan 分支先阻塞 +// 排队（sticky 45s / fallback 30s）拿到槽位才终检，无上限重选会把单次请求的 +// 延迟放大到 N × WaitPlan.Timeout。 +func recordOpenAIProfitVeto(failedAccountIDs map[int64]struct{}, accountID int64, vetoCount *int) bool { + failedAccountIDs[accountID] = struct{}{} + *vetoCount++ + return *vetoCount < maxProfitVetoAttempts +} + +// handleOpenAIProfitVetoExhausted 在利润否决预算耗尽时写出错误响应。 +// 与 acquireResponsesAccountSlot 内部的 no-available-accounts 失败分支同形， +// 保证同一调用方在两条路径上拿到一致的响应格式。 +func (h *OpenAIGatewayHandler) handleOpenAIProfitVetoExhausted( + c *gin.Context, + streamStarted bool, + reqLog *zap.Logger, + vetoCount int, +) { + reqLog.Warn("openai.profit_veto_attempts_exhausted", zap.Int("profit_veto_count", vetoCount)) + markOpsRoutingCapacityLimited(c) + h.handleStreamingAwareError(c, http.StatusServiceUnavailable, "api_error", profitVetoExhaustedMessage, streamStarted) +} + func (h *OpenAIGatewayHandler) acquireResponsesAccountSlot( c *gin.Context, groupID *int64, @@ -1747,6 +1812,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { ) maxAccountSwitches := h.maxAccountSwitches switchCount := 0 + profitVetoCount := 0 failedAccountIDs := make(map[int64]struct{}) var lastFailoverErr *service.UpstreamFailoverError var oauth429FailoverState service.OpenAIOAuth429FailoverState @@ -1801,7 +1867,8 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { // ctx）。连接内不重选号，但每个 turn 开始经 BeforeTurn 重新冻结 pricingAt // 并按最新门复核当前账号（准入与计费同源），峰前建连保活不能让后续 turn // 继续按建连时刻的谷价计费。生图意图只影响能力路由与图片计费，不关门。 - wsPricingCtx, wsPricingAt := h.gatewayService.WithOpenAIRequestPricingContext(ctx, apiKey.GroupID) + // 建连时刻只用于选号/准入，不作为任何 turn 的计费定价时刻。 + wsPricingCtx, _ := h.gatewayService.WithOpenAIRequestPricingContext(ctx, apiKey.GroupID) ctx = wsPricingCtx for { @@ -1861,7 +1928,11 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { accountReleaseFunc() } reqLog.Debug("openai.websocket_account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) - failedAccountIDs[account.ID] = struct{}{} + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + reqLog.Warn("openai.websocket_profit_veto_attempts_exhausted", zap.Int("profit_veto_count", profitVetoCount)) + closeOpenAIClientWS(wsConn, coderws.StatusTryAgainLater, "no available account") + return + } continue } account = latest @@ -1894,7 +1965,11 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { fastReleaseFunc() } reqLog.Debug("openai.websocket_account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason)) - failedAccountIDs[account.ID] = struct{}{} + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + reqLog.Warn("openai.websocket_profit_veto_attempts_exhausted", zap.Int("profit_veto_count", profitVetoCount)) + closeOpenAIClientWS(wsConn, coderws.StatusTryAgainLater, "no available account") + return + } continue } account = latest @@ -1944,10 +2019,9 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { var turnChannelMapping atomic.Pointer[openAIWSTurnChannelMappingSnapshot] turnChannelMapping.Store(&openAIWSTurnChannelMappingSnapshot{turn: 1, mapping: channelMappingWS}) // turn 级定价：BeforeTurn 重新冻结 pricingAt 并按最新门复核当前账号， - // AfterTurn 的计费读取所属 turn 的时刻；turn 在连接内串行推进，互斥锁 - // 只为跨用量提交 goroutine 的读取安全。 - var turnPricingMu sync.Mutex - turnPricingAt := wsPricingAt + // AfterTurn 的计费读取所属 turn 的时刻。零值起步的语义见 + // openAIWSTurnPricing 的注释——绝不能用建连时刻初始化。 + var turnPricing openAIWSTurnPricing hooks := &service.OpenAIWSIngressHooks{ InitialRequestModel: reqModel, MaxReasoningEffort: maxReasoningEffort, @@ -2004,9 +2078,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { zap.String("reason", reason)) return service.NewOpenAIWSClientCloseError(coderws.StatusTryAgainLater, "account is no longer eligible for this connection, please reconnect", nil) } - turnPricingMu.Lock() - turnPricingAt = turnAt - turnPricingMu.Unlock() + turnPricing.freeze(turnAt) if turn == 1 { return nil } @@ -2105,9 +2177,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { upstreamEndpoint := resolveOpenAIUpstreamEndpoint(c, account, result) quotaPlatform := service.QuotaPlatform(c.Request.Context(), apiKey) sessionID := service.ExtractClientSessionID(c) - turnPricingMu.Lock() - turnRecordPricingAt := turnPricingAt - turnPricingMu.Unlock() + turnRecordPricingAt := turnPricing.current() cyberBlocked := service.GetOpsCyberPolicy(c) != nil h.submitOpenAIUsageRecordTask(ctx, result, func(taskCtx context.Context) { if err := h.gatewayService.RecordUsage(taskCtx, &service.OpenAIRecordUsageInput{ diff --git a/backend/internal/handler/openai_images.go b/backend/internal/handler/openai_images.go index 721c206981..d5d16d1e71 100644 --- a/backend/internal/handler/openai_images.go +++ b/backend/internal/handler/openai_images.go @@ -148,6 +148,7 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) { maxAccountSwitches := h.maxAccountSwitches switchCount := 0 + profitVetoCount := 0 failedAccountIDs := make(map[int64]struct{}) sameAccountRetryCount := make(map[int64]int) var lastFailoverErr *service.UpstreamFailoverError @@ -223,8 +224,11 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) { accountReleaseFunc, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, parsed.Stream, &streamStarted, reqLog) if slotResult == openAISlotAcquireProfitVetoed { - // Images 调度不装利润门，此分支实际不可达；防御性排除重选。 - failedAccountIDs[account.ID] = struct{}{} + // Images 调度不装利润门，此分支实际不可达；防御性排除重选并受同一否决上限约束。 + if !recordOpenAIProfitVeto(failedAccountIDs, account.ID, &profitVetoCount) { + h.handleOpenAIProfitVetoExhausted(c, streamStarted, reqLog, profitVetoCount) + return + } continue } if slotResult != openAISlotAcquireOK { diff --git a/backend/internal/handler/openai_profit_veto_budget_test.go b/backend/internal/handler/openai_profit_veto_budget_test.go new file mode 100644 index 0000000000..60e4871121 --- /dev/null +++ b/backend/internal/handler/openai_profit_veto_budget_test.go @@ -0,0 +1,74 @@ +package handler + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +// TestRecordOpenAIProfitVetoBounded 钉死 OpenAI 侧选号循环的利润否决预算： +// 账号被加入排除集，且第 maxProfitVetoAttempts 次否决返回 false 要求终止。 +// +// 上限是必需的：acquireResponsesAccountSlot 的 WaitPlan 分支先阻塞排队 +// （sticky 45s / fallback 30s）拿到槽位才做利润终检，无上限重选会把单次请求 +// 的延迟放大到 N × WaitPlan.Timeout。 +func TestRecordOpenAIProfitVetoBounded(t *testing.T) { + failed := make(map[int64]struct{}) + count := 0 + + for i := int64(1); i < int64(maxProfitVetoAttempts); i++ { + require.True(t, recordOpenAIProfitVeto(failed, i, &count), "第 %d 次否决应允许继续重选", i) + require.Contains(t, failed, i, "否决的账号必须进入本请求排除集") + } + + require.False(t, + recordOpenAIProfitVeto(failed, int64(maxProfitVetoAttempts), &count), + "第 %d 次否决应耗尽预算并要求终止", maxProfitVetoAttempts) + require.Equal(t, maxProfitVetoAttempts, count) + require.Len(t, failed, maxProfitVetoAttempts) +} + +// TestOpenAIProfitVetoLoopTerminates 模拟「选号 → 抢槽 → 利润终检否决 → 重选」 +// 循环：整池越线时必须在常数步内终止，而不是把候选池逐个排队一遍。 +func TestOpenAIProfitVetoLoopTerminates(t *testing.T) { + failed := make(map[int64]struct{}) + count := 0 + terminated := false + iterations := 0 + + // 候选池远大于否决上限：无上限时会对每个账号各排队一次。 + for accountID := int64(1); accountID <= 500; accountID++ { + iterations++ + if _, excluded := failed[accountID]; excluded { + continue + } + if !recordOpenAIProfitVeto(failed, accountID, &count) { + terminated = true + break + } + } + + require.True(t, terminated, "整池越线时循环必须由否决预算终止") + require.Equal(t, maxProfitVetoAttempts, iterations) +} + +// TestProfitVetoBudgetSharedWithFailoverState 钉死两条路径（FailoverState 与 +// OpenAI 独立计数器）使用同一上限语义，避免日后单边漂移。 +func TestProfitVetoBudgetSharedWithFailoverState(t *testing.T) { + fs := NewFailoverState(10, false) + failed := make(map[int64]struct{}) + count := 0 + + var fsStoppedAt, openAIStoppedAt int + for i := int64(1); i <= int64(maxProfitVetoAttempts)+5; i++ { + if fsStoppedAt == 0 && fs.RecordProfitVeto(i) == FailoverExhausted { + fsStoppedAt = fs.ProfitVetoCount() + } + if openAIStoppedAt == 0 && !recordOpenAIProfitVeto(failed, i, &count) { + openAIStoppedAt = count + } + } + + require.Equal(t, maxProfitVetoAttempts, fsStoppedAt) + require.Equal(t, fsStoppedAt, openAIStoppedAt, "两条路径的利润否决上限必须一致") +} diff --git a/backend/internal/handler/openai_ws_turn_pricing_test.go b/backend/internal/handler/openai_ws_turn_pricing_test.go new file mode 100644 index 0000000000..b9bb204fd6 --- /dev/null +++ b/backend/internal/handler/openai_ws_turn_pricing_test.go @@ -0,0 +1,34 @@ +package handler + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestOpenAIWSTurnPricingZeroValue 钉死 WS turn 定价的零值语义： +// 没有 turn 起始回调的 ingress 模式（ws_v2 passthrough 只实现 AfterTurn） +// 必须让 pricingAt 保持零，由 RecordUsage 回退到记录时刻。 +// +// 反例（本 PR 引入的回归）：用建连时刻初始化，会把透传连接的所有 turn 钉死在 +// 建连时的高峰因子——客户端峰前一分钟建连并保活，整条连接就按谷价结算。 +func TestOpenAIWSTurnPricingZeroValue(t *testing.T) { + var p openAIWSTurnPricing + require.True(t, p.current().IsZero(), + "未经 turn 起始回调冻结时必须保持零值，交由 RecordUsage 回退记录时刻") +} + +// TestOpenAIWSTurnPricingFreezePerTurn 钉死每个 turn 的 BeforeTurn 都会覆盖 +// 上一个 turn 的定价时刻：长连接跨峰谷时后续 turn 不得沿用旧时刻。 +func TestOpenAIWSTurnPricingFreezePerTurn(t *testing.T) { + var p openAIWSTurnPricing + turn1 := time.Now().Add(-time.Hour) + turn2 := time.Now() + + p.freeze(turn1) + require.Equal(t, turn1, p.current()) + + p.freeze(turn2) + require.Equal(t, turn2, p.current(), "后续 turn 必须使用自己的定价时刻") +} diff --git a/backend/internal/repository/scheduler_cache_rate_multiplier_unit_test.go b/backend/internal/repository/scheduler_cache_rate_multiplier_unit_test.go new file mode 100644 index 0000000000..fe03a65634 --- /dev/null +++ b/backend/internal/repository/scheduler_cache_rate_multiplier_unit_test.go @@ -0,0 +1,81 @@ +//go:build unit + +package repository + +import ( + "context" + "testing" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" +) + +// TestSchedulerCachePreservesRateMultiplier 钉死账号调度快照的两份 payload +// （full + metadata）都必须保留 RateMultiplier。 +// +// 为什么值得一条专门的护栏：分组利润门对 nil RateMultiplier 是 fail-closed +// （保守拒绝），与全系统其他地方不同——Account.BillingRateMultiplier() 对同样 +// 的 nil 返回 1.0，字段注释也写着「nil 表示按 1.0 处理」。DB 列有 Default(1.0) +// 且非 nillable，因此 nil 只可能来自缓存反序列化缺字段。 +// buildSchedulerMetadataAccount 是显式字段清单，将来任何一次快照结构调整漏掉 +// 这一项，对启用利润控制的分组就是全组 no available accounts——而且是静默的。 +// 这条测试让那种漏列在 CI 就变红。 +func TestSchedulerCachePreservesRateMultiplier(t *testing.T) { + rate := 0.75 + account := service.Account{ + ID: 9001, + Name: "profit-gate-rate", + Platform: service.PlatformOpenAI, + Type: service.AccountTypeAPIKey, + Status: service.StatusActive, + Schedulable: true, + Concurrency: 2, + RateMultiplier: &rate, + } + + t.Run("metadata payload keeps the field", func(t *testing.T) { + meta := buildSchedulerMetadataAccount(account) + require.NotNil(t, meta.RateMultiplier, "metadata 快照漏掉 rate_multiplier 会让利润门 fail-closed 全组拒绝") + require.Equal(t, rate, *meta.RateMultiplier) + }) + + t.Run("both payloads survive a decode round-trip", func(t *testing.T) { + full, meta, err := marshalSchedulerCacheAccount(account) + require.NoError(t, err) + + for name, payload := range map[string][]byte{"full": full, "metadata": meta} { + decoded, decodeErr := decodeCachedAccount(payload) + require.NoError(t, decodeErr, name) + require.NotNil(t, decoded.RateMultiplier, "%s payload 反序列化后 rate_multiplier 不得为 nil", name) + require.Equal(t, rate, *decoded.RateMultiplier, name) + } + }) + + t.Run("zero rate survives as zero rather than nil", func(t *testing.T) { + // 0 是合法值（该账号上游成本为 0），绝不能被当成缺字段丢掉： + // 丢成 nil 后利润门会把一个本该必然放行的账号判成越线。 + zero := 0.0 + zeroAccount := account + zeroAccount.ID = 9002 + zeroAccount.RateMultiplier = &zero + + _, meta, err := marshalSchedulerCacheAccount(zeroAccount) + require.NoError(t, err) + decoded, err := decodeCachedAccount(meta) + require.NoError(t, err) + require.NotNil(t, decoded.RateMultiplier) + require.Equal(t, 0.0, *decoded.RateMultiplier) + }) + + t.Run("SetAccount then GetAccount preserves the field", func(t *testing.T) { + cache := newSchedulerCacheUnit(t) + ctx := context.Background() + require.NoError(t, cache.SetAccount(ctx, &account)) + + got, err := cache.GetAccount(ctx, account.ID) + require.NoError(t, err) + require.NotNil(t, got) + require.NotNil(t, got.RateMultiplier, "端到端缓存读写后 rate_multiplier 不得丢失") + require.Equal(t, rate, *got.RateMultiplier) + }) +} diff --git a/backend/internal/server/api_contract_test.go b/backend/internal/server/api_contract_test.go index 5b09d42442..f0ad8dcdb3 100644 --- a/backend/internal/server/api_contract_test.go +++ b/backend/internal/server/api_contract_test.go @@ -314,19 +314,23 @@ func TestAPIContracts(t *testing.T) { name: "GET /api/v1/groups/available", setup: func(t *testing.T, deps *contractDeps) { t.Helper() - // 普通用户可见的分组列表不应包含内部字段（如 model_routing/account_count）。 + // 普通用户可见的分组列表不应包含内部字段（如 model_routing/account_count）， + // 也不得包含利润控制配置——它与同响应的 rate_multiplier 相乘即可反推上游成本上限。 deps.groupRepo.SetActive([]service.Group{ { - ID: 10, - Name: "Group One", - Description: "desc", - Platform: service.PlatformAnthropic, - RateMultiplier: 1.5, - PeakRateMultiplier: 1.0, - IsExclusive: false, - Status: service.StatusActive, - SubscriptionType: service.SubscriptionTypeStandard, - ModelRoutingEnabled: true, + ID: 10, + Name: "Group One", + Description: "desc", + Platform: service.PlatformAnthropic, + RateMultiplier: 1.5, + PeakRateMultiplier: 1.0, + IsExclusive: false, + Status: service.StatusActive, + SubscriptionType: service.SubscriptionTypeStandard, + ProfitControlEnabled: true, + ProfitMinMargin: 0.3, + ProfitSafetyBuffer: 0.05, + ModelRoutingEnabled: true, ModelRouting: map[string][]int64{ "claude-3-*": []int64{101, 102}, }, @@ -354,9 +358,6 @@ func TestAPIContracts(t *testing.T) { "peak_start": "", "peak_end": "", "peak_rate_multiplier": 1, - "profit_control_enabled": false, - "profit_min_margin": 0, - "profit_safety_buffer": 0, "is_exclusive": false, "status": "active", "subscription_type": "standard", diff --git a/backend/internal/service/admin_group.go b/backend/internal/service/admin_group.go index 2d4a3f7fcb..910f668438 100644 --- a/backend/internal/service/admin_group.go +++ b/backend/internal/service/admin_group.go @@ -300,10 +300,7 @@ func (s *adminServiceImpl) CreateGroup(ctx context.Context, input *CreateGroupIn return nil, errors.New("rate_multiplier must be > 0") } - platform := input.Platform - if platform == "" { - platform = PlatformAnthropic - } + platform := NormalizeGroupPlatform(input.Platform) maxReasoningEffort, err := normalizeMaxReasoningEffortForPlatform(platform, input.MaxReasoningEffort) if err != nil { return nil, infraerrors.Newf(http.StatusBadRequest, "INVALID_MAX_REASONING_EFFORT", "%v", err) diff --git a/backend/internal/service/api_key_auth_cache.go b/backend/internal/service/api_key_auth_cache.go index a6f76beb80..a665cdad3c 100644 --- a/backend/internal/service/api_key_auth_cache.go +++ b/backend/internal/service/api_key_auth_cache.go @@ -115,9 +115,13 @@ type APIKeyAuthGroupSnapshot struct { PeakEnd string `json:"peak_end"` PeakRateMultiplier float64 `json:"peak_rate_multiplier"` - // 分组利润控制：调度准入门按 schedulerSnapshot 实时读取分组配置， - // 不依赖本快照；此处随快照缓存只为保证 apiKey.Group 字段完整， - // 任何消费方都不会读到误导性的零值。 + // 分组利润控制：调度准入门在直连热路径上读的就是这份快照——门解析 + // （resolveOpenAIProfitControlGate / resolveProfitControlGroup）优先取 + // 认证中间件放入 ctx 的 Group，而它正是本快照物化出来的对象，生产绝大 + // 多数流量走的都是这条路；只有 composite/模型路由等被调度分组与认证分组 + // 不一致时才回源 schedulerSnapshot。 + // 因此这三个字段与 GetByKeyForAuth 的投影都不得删减：漏掉任何一个， + // 门会拿到零值 ProfitControlEnabled=false 而静默失效（有集成测试兜底）。 ProfitControlEnabled bool `json:"profit_control_enabled"` ProfitMinMargin float64 `json:"profit_min_margin"` ProfitSafetyBuffer float64 `json:"profit_safety_buffer"` diff --git a/backend/internal/service/gateway_profit_control.go b/backend/internal/service/gateway_profit_control.go index c2888a1b45..fe8ba14752 100644 --- a/backend/internal/service/gateway_profit_control.go +++ b/backend/internal/service/gateway_profit_control.go @@ -3,7 +3,6 @@ package service import ( "context" "log/slog" - "math" "github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey" ) @@ -43,10 +42,7 @@ func (s *GatewayService) withGatewayProfitControlGate(ctx context.Context, group downstream = s.ResolveUserGroupRateMultiplier(ctx, userID, billingGroup.ID, billingGroup.RateMultiplier) } downstream *= billingGroup.PeakMultiplierAt(pricingAt) - threshold := downstream * (1 - group.ProfitMinMargin - group.ProfitSafetyBuffer) - if math.IsNaN(threshold) || math.IsInf(threshold, 0) || threshold < 0 { - threshold = 0 - } + threshold := clampProfitControlThreshold(downstream * (1 - group.ProfitMinMargin - group.ProfitSafetyBuffer)) gate := &openAIProfitControlGate{ groupID: group.ID, @@ -71,7 +67,8 @@ func (s *GatewayService) resolveProfitControlGroup(ctx context.Context, groupID return group, nil } if s.schedulerSnapshot != nil { - return s.schedulerSnapshot.GetGroupByID(ctx, groupID) + // Lite 读取：门只用平台/倍率/利润/高峰字段，不需要账号计数聚合。 + return s.schedulerSnapshot.GetGroupByIDLite(ctx, groupID) } return s.resolveGroupByID(ctx, groupID) } diff --git a/backend/internal/service/gateway_profit_control_v2_test.go b/backend/internal/service/gateway_profit_control_v2_test.go index 312ade0fd2..c05f4501e1 100644 --- a/backend/internal/service/gateway_profit_control_v2_test.go +++ b/backend/internal/service/gateway_profit_control_v2_test.go @@ -148,10 +148,15 @@ type profitControlFailingGroupRepo struct { GroupRepository } -func (profitControlFailingGroupRepo) GetByID(context.Context, int64) (*Group, error) { +func (profitControlFailingGroupRepo) GetByIDLite(context.Context, int64) (*Group, error) { return nil, errors.New("group cache unavailable") } +// 见 profitControlGroupRepo.GetByID：利润门必须走不带账号计数聚合的 lite 读取。 +func (profitControlFailingGroupRepo) GetByID(context.Context, int64) (*Group, error) { + panic("profit control gate must read groups via GetByIDLite (no account-count aggregation)") +} + func TestGatewayProfitControlLegacyMixedAndRoutedSelection(t *testing.T) { t.Run("legacy single-platform selection", func(t *testing.T) { group := gatewayProfitTestGroup(111, PlatformGrok) diff --git a/backend/internal/service/group.go b/backend/internal/service/group.go index 549a37e40c..2fea013aff 100644 --- a/backend/internal/service/group.go +++ b/backend/internal/service/group.go @@ -351,6 +351,17 @@ func validProfitControlRatio(v float64) bool { return !math.IsNaN(v) && !math.IsInf(v, 0) && v >= 0 && v < 1 } +// NormalizeGroupPlatform 把创建分组时省略的 platform 归一化为默认平台。 +// handler 的入参预校验必须与 CreateGroup 落库时用同一个归一化结果，否则 +// 「省略 platform + 启用利润控制」会被 handler 以「平台不支持」400 掉， +// 而该分组本会被建成受支持的 anthropic 分组。 +func NormalizeGroupPlatform(platform string) string { + if platform == "" { + return PlatformAnthropic + } + return platform +} + // ValidateProfitControlConfig 是分组利润控制配置的唯一校验来源，handler 与 service 层共用。 // enabled=true 时仅允许五个可计费平台分组；margin/buffer 各自 ∈ [0,1)，且 margin+buffer < 1 // （相加 >=1 时阈值 <=0，所有可核价账号都会被排除，视为配置错误而不是静默全黑）。 diff --git a/backend/internal/service/group_profit_platform_test.go b/backend/internal/service/group_profit_platform_test.go new file mode 100644 index 0000000000..28a5f2290a --- /dev/null +++ b/backend/internal/service/group_profit_platform_test.go @@ -0,0 +1,20 @@ +package service + +import "testing" + +// TestNormalizeGroupPlatformDefaultsToAnthropic 钉死创建分组时省略 platform 的 +// 归一化结果：handler 的利润控制预校验与 CreateGroup 落库共用这个函数， +// 两边一旦漂移，「省略 platform + profit_control_enabled=true」的创建请求 +// 会被 handler 以「平台不支持」400 掉，而该分组本会建成受支持的 anthropic。 +func TestNormalizeGroupPlatformDefaultsToAnthropic(t *testing.T) { + if got := NormalizeGroupPlatform(""); got != PlatformAnthropic { + t.Fatalf("空 platform 应归一化为 %s，got %s", PlatformAnthropic, got) + } + if got := NormalizeGroupPlatform(PlatformOpenAI); got != PlatformOpenAI { + t.Fatalf("显式 platform 不应被改写，got %s", got) + } + // 归一化后的默认平台必须真的支持利润控制，否则预校验仍会误拒。 + if err := ValidateProfitControlConfig(NormalizeGroupPlatform(""), true, 0.3, 0.05); err != nil { + t.Fatalf("省略 platform 且启用利润控制不应被预校验拒绝: %v", err) + } +} diff --git a/backend/internal/service/openai_profit_control.go b/backend/internal/service/openai_profit_control.go index fa76dfa826..4153bc4094 100644 --- a/backend/internal/service/openai_profit_control.go +++ b/backend/internal/service/openai_profit_control.go @@ -1,6 +1,6 @@ package service -// 分组利润控制（配套 migration 191/192 的 groups.profit_* 字段）。 +// 分组利润控制（配套 migration 192/193 的 groups.profit_* 字段）。 // // 定位：利润控制是"候选准入过滤"，只决定账号能否进入调度候选池；既有的排序、 // 评分、粘性、熔断、负载均衡在合格账号之间照常工作，本文件不改变它们的行为。 @@ -96,6 +96,23 @@ type openAIProfitControlSuppressCtxKey struct{} // 的高峰因子共用，保证一个请求从准入到扣费不中途变价。 type openAIPricingAtCtxKey struct{} +// clampProfitControlThreshold 归一化利润门阈值。Validate/Normalize 已保证 +// margin+buffer < 1，这里只对存量脏数据兜底：阈值非有限或为负时按 0 处理 +// （等价于只放行免费上游）。装门点与 profit-preview 共用，避免口径漂移。 +func clampProfitControlThreshold(threshold float64) float64 { + if math.IsNaN(threshold) || math.IsInf(threshold, 0) || threshold < 0 { + return 0 + } + return threshold +} + +// profitControlOverThreshold 是"上游倍率越线"的唯一定义：相对 epsilon 吸收 +// decimal(10,4) 落库与浮点乘法的边界误差，U == 阈值 判定为合格。 +// 线上否决点与 profit-preview 共用，两者不得各自实现。 +func profitControlOverThreshold(upstream, threshold float64) bool { + return upstream-threshold > profitControlRateEpsilon*math.Max(1, math.Abs(threshold)) +} + // openAIProfitControlGate 是一个请求的利润准入门。除 pricingAt 外全部为预计算 // 标量：候选过滤热路径上每账号只做一次快照解码与一次浮点比较。 type openAIProfitControlGate struct { @@ -209,7 +226,8 @@ func (s *OpenAIGatewayService) resolveOpenAIProfitControlGate(ctx context.Contex if ctxGroup, ok := ctx.Value(ctxkey.Group).(*Group); ok && IsGroupContextValid(ctxGroup) && ctxGroup.ID == *groupID { group = ctxGroup } else if s.schedulerSnapshot != nil { - loaded, err := s.schedulerSnapshot.GetGroupByID(ctx, *groupID) + // Lite 读取：门只用平台/倍率/利润/高峰字段，不需要账号计数聚合。 + loaded, err := s.schedulerSnapshot.GetGroupByIDLite(ctx, *groupID) if err != nil { // fail-open：配置系统故障时可用性优先，该窗口内利润保证不成立， // 依赖 WARN 暴露；不把瞬时 DB 抖动放大成全站不可调度。 @@ -242,12 +260,7 @@ func (s *OpenAIGatewayService) resolveOpenAIProfitControlGate(ctx context.Contex downstream *= billingGroup.PeakMultiplierAt(pricingAt) deduction := group.ProfitMinMargin + group.ProfitSafetyBuffer - threshold := downstream * (1 - deduction) - // Validate/Normalize 已保证 margin+buffer < 1；此处仅对存量脏数据兜底， - // 阈值非有限或为负时按 0 处理（等价于只放行免费上游）。 - if math.IsNaN(threshold) || math.IsInf(threshold, 0) || threshold < 0 { - threshold = 0 - } + threshold := clampProfitControlThreshold(downstream * (1 - deduction)) return &openAIProfitControlGate{ groupID: *groupID, platform: group.Platform, @@ -299,7 +312,7 @@ func openAIProfitControlVetoReason(ctx context.Context, account *Account) (bool, return true, openAIProfitFilterReasonInvalidAccountRate } upstream := *account.RateMultiplier - if upstream-gate.threshold > profitControlRateEpsilon*math.Max(1, math.Abs(gate.threshold)) { + if profitControlOverThreshold(upstream, gate.threshold) { openAIProfitControlObserverInstance.recordVeto(gate.groupID, gate.platform, gate.threshold, openAIProfitFilterReasonThreshold) return true, openAIProfitFilterReasonThreshold } diff --git a/backend/internal/service/openai_profit_control_paths_test.go b/backend/internal/service/openai_profit_control_paths_test.go index 4c02c887bb..239a7d26f2 100644 --- a/backend/internal/service/openai_profit_control_paths_test.go +++ b/backend/internal/service/openai_profit_control_paths_test.go @@ -266,10 +266,17 @@ type profitControlGroupRepo struct { group *Group } -func (r profitControlGroupRepo) GetByID(context.Context, int64) (*Group, error) { +func (r profitControlGroupRepo) GetByIDLite(context.Context, int64) (*Group, error) { return r.group, nil } +// GetByID 故意 panic：利润门只需要分组配置，不需要 GetByID 附带的账号计数 +// 聚合查询。装门走 GetByID 会在 composite/模型路由/fallback 的每次装门（WS +// 每 turn 一次）上多打一条聚合，且发生在「是否启用利润控制」判定之前。 +func (r profitControlGroupRepo) GetByID(context.Context, int64) (*Group, error) { + panic("profit control gate must read groups via GetByIDLite (no account-count aggregation)") +} + // composite 路由：门配置取被调度成员分组，D 取请求真实计费分组（ctx 认证分组）。 func TestProfitControl_CompositeUsesBillingGroupRate(t *testing.T) { memberGroupID := int64(7) diff --git a/backend/internal/service/openai_ws_forwarder_ingress.go b/backend/internal/service/openai_ws_forwarder_ingress.go index 3664ae8b20..23582c2cea 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress.go +++ b/backend/internal/service/openai_ws_forwarder_ingress.go @@ -77,6 +77,11 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( if wsDecision.Transport != OpenAIUpstreamTransportResponsesWebsocketV2 { return fmt.Errorf("websocket ingress requires ws_v2 transport, got=%s", wsDecision.Transport) } + // 注意：透传 relay 只回调 hooks.AfterTurn，没有 turn 起始回调， + // 因此下面这条路径永远不会触发 hooks.BeforeTurn——分组利润控制的 + // turn 级复核与 turn 级 pricingAt 冻结都不覆盖透传 ingress， + // 只有建连时的准入门生效。handler 侧据此把 turn 定价留作零值， + // 由 RecordUsage 回退到记录时刻（见 openAIWSTurnPricing 注释）。 return s.proxyResponsesWebSocketV2Passthrough( ctx, c, diff --git a/backend/internal/service/openai_ws_passthrough_turn_pricing_test.go b/backend/internal/service/openai_ws_passthrough_turn_pricing_test.go new file mode 100644 index 0000000000..bb6e5ebde6 --- /dev/null +++ b/backend/internal/service/openai_ws_passthrough_turn_pricing_test.go @@ -0,0 +1,128 @@ +package service + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + coderws "github.com/coder/websocket" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "github.com/tidwall/gjson" +) + +// startPassthroughHookRecordingServer 与 startPassthroughLifecycleServer 相同， +// 但把一组会记录调用的 hooks 传给 ingress，用于观察透传路径的 turn 回调。 +func startPassthroughHookRecordingServer( + t *testing.T, + controlCtx context.Context, + svc *OpenAIGatewayService, + account *Account, + hooks *OpenAIWSIngressHooks, +) (*httptest.Server, <-chan error) { + t.Helper() + serverErr := make(chan error, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := coderws.Accept(w, r, &coderws.AcceptOptions{CompressionMode: coderws.CompressionContextTakeover}) + if err != nil { + serverErr <- err + return + } + defer func() { _ = conn.CloseNow() }() + + msgType, firstMessage, err := ReadOpenAIWSClientMessage( + controlCtx, + conn, + 3*time.Second, + coderws.StatusPolicyViolation, + "missing first response.create message", + ) + if err != nil { + serverErr <- err + return + } + if msgType != coderws.MessageText { + serverErr <- errors.New("first message was not text") + return + } + + recorder := httptest.NewRecorder() + ginCtx, _ := gin.CreateTestContext(recorder) + req := r.Clone(controlCtx) + req.Header = req.Header.Clone() + ginCtx.Request = req + serverErr <- svc.ProxyResponsesWebSocketFromClient(controlCtx, ginCtx, conn, account, "sk-test", firstMessage, hooks) + })) + return server, serverErr +} + +// TestPassthroughIngressNeverCallsBeforeTurn 钉死 ws_v2 透传 ingress 与 handler +// 侧 turn 定价的耦合：透传 relay 只回调 AfterTurn，没有任何 turn 起始回调， +// 因此 hooks.BeforeTurn 永远不会触发。 +// +// handler 依赖这一点：openAIWSTurnPricing 零值起步，透传连接的每个 turn 都拿 +// 不到冻结的 pricingAt，RecordUsage 回退到记录时刻——与引入分组利润控制前的 +// 基线一致。若把 turn 定价初始化成建连时刻，透传连接的所有 turn 就会被钉死在 +// 建连时的高峰因子，客户端峰前建连保活即可全程按谷价结算。 +// +// 若本断言因为透传补齐了 turn 起始回调而失败：这是好事，请同步复核 +// openAIWSTurnPricing 的零值语义与透传路径的 turn 级利润复核。 +func TestPassthroughIngressNeverCallsBeforeTurn(t *testing.T) { + gin.SetMode(gin.TestMode) + controlCtx, cancelControl := context.WithCancelCause(context.Background()) + defer cancelControl(context.Canceled) + + upstream := newStagedPassthroughConn() + upstream.Send(`{"type":"response.completed","response":{"id":"resp_pricing","model":"gpt-5.1","usage":{"input_tokens":1,"output_tokens":1}}}`) + + var hooksMu sync.Mutex + beforeTurnCalls := 0 + afterTurnCalls := 0 + hooks := &OpenAIWSIngressHooks{ + BeforeTurn: func(int) error { + hooksMu.Lock() + beforeTurnCalls++ + hooksMu.Unlock() + return nil + }, + AfterTurn: func(int, *OpenAIForwardResult, error) { + hooksMu.Lock() + afterTurnCalls++ + hooksMu.Unlock() + }, + } + + server, serverErr := startPassthroughHookRecordingServer( + t, + controlCtx, + newPassthroughLifecycleService(passthroughLifecycleConfig(), upstream), + passthroughLifecycleAccount(), + hooks, + ) + defer server.Close() + clientConn := dialPassthroughLifecycleClient(t, server) + defer func() { _ = clientConn.CloseNow() }() + + event, err := readPassthroughLifecycleFrame(t, clientConn, 3*time.Second) + require.NoError(t, err) + require.Equal(t, "response.completed", gjson.GetBytes(event, "type").String()) + + // 等待连接自然结束（inter-turn idle 超时），确保 AfterTurn 已提交。 + _, _ = readPassthroughLifecycleFrame(t, clientConn, 3*time.Second) + select { + case <-serverErr: + case <-time.After(3 * time.Second): + t.Fatal("passthrough ingress did not exit") + } + + hooksMu.Lock() + gotBefore, gotAfter := beforeTurnCalls, afterTurnCalls + hooksMu.Unlock() + + require.Zero(t, gotBefore, "透传 ingress 没有 turn 起始回调，BeforeTurn 不应被调用") + require.Positive(t, gotAfter, "透传 ingress 仍应回调 AfterTurn 提交用量") +} diff --git a/backend/internal/service/profit_control_threshold_helpers_test.go b/backend/internal/service/profit_control_threshold_helpers_test.go new file mode 100644 index 0000000000..15f5bc1838 --- /dev/null +++ b/backend/internal/service/profit_control_threshold_helpers_test.go @@ -0,0 +1,38 @@ +package service + +import ( + "math" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestClampProfitControlThreshold 钉死阈值清洗口径：非有限或为负一律按 0 +// （等价于只放行免费上游），合法值原样保留。 +func TestClampProfitControlThreshold(t *testing.T) { + require.Equal(t, 0.0, clampProfitControlThreshold(math.NaN())) + require.Equal(t, 0.0, clampProfitControlThreshold(math.Inf(1))) + require.Equal(t, 0.0, clampProfitControlThreshold(math.Inf(-1))) + require.Equal(t, 0.0, clampProfitControlThreshold(-0.5)) + require.Equal(t, 0.0, clampProfitControlThreshold(0)) + require.Equal(t, 0.7, clampProfitControlThreshold(0.7)) +} + +// TestProfitControlOverThreshold 钉死「越线」的唯一定义：相对 epsilon 吸收 +// decimal(10,4) 落库与浮点乘法的边界误差，U == 阈值 判定为合格。 +// +// 线上否决点（openAIProfitControlVetoReason）与 profit-preview 现在共用这个 +// 函数与 clampProfitControlThreshold；此前各复制了一份，会随线上口径漂移， +// 导致预演结果与真实调度不一致。 +func TestProfitControlOverThreshold(t *testing.T) { + require.False(t, profitControlOverThreshold(0.7, 0.7), "U == 阈值 必须判定为合格") + require.False(t, profitControlOverThreshold(0.6999, 0.7)) + require.True(t, profitControlOverThreshold(0.7001, 0.7)) + + // 0.3*3 的浮点尾数不得被判成越线。 + require.False(t, profitControlOverThreshold(0.1*3, 0.3)) + + // 阈值为 0 时只放行 0 成本上游。 + require.False(t, profitControlOverThreshold(0, 0)) + require.True(t, profitControlOverThreshold(0.0001, 0)) +} diff --git a/backend/internal/service/profit_preview.go b/backend/internal/service/profit_preview.go index 5cb8a78a89..d63b9317ce 100644 --- a/backend/internal/service/profit_preview.go +++ b/backend/internal/service/profit_preview.go @@ -115,8 +115,8 @@ func PreviewProfitAdmission(inputs []ProfitPreviewGroupInput, evalAt time.Time) } minD := minRate * peak deduction := group.ProfitMinMargin + group.ProfitSafetyBuffer - thresholdDefault := clampProfitPreviewThreshold(defaultD * (1 - deduction)) - thresholdMinD := clampProfitPreviewThreshold(minD * (1 - deduction)) + thresholdDefault := clampProfitControlThreshold(defaultD * (1 - deduction)) + thresholdMinD := clampProfitControlThreshold(minD * (1 - deduction)) report.DefaultD = defaultD report.MinEffectiveD = minD report.ThresholdDefault = thresholdDefault @@ -183,11 +183,11 @@ func previewAccountProfitAdmission( verdict.Class = ProfitPreviewClassAdmitted case !validRate: verdict.Class = ProfitPreviewClassRejectedInvalidRate - case profitPreviewOverThreshold(*account.RateMultiplier, thresholdDefault): + case profitControlOverThreshold(*account.RateMultiplier, thresholdDefault): verdict.Class = ProfitPreviewClassRejectedThreshold default: verdict.Class = ProfitPreviewClassAdmitted - verdict.RejectedUnderMinD = profitPreviewOverThreshold(*account.RateMultiplier, thresholdMinD) + verdict.RejectedUnderMinD = profitControlOverThreshold(*account.RateMultiplier, thresholdMinD) } return verdict } @@ -209,15 +209,3 @@ func profitPreviewProbeWarnings(account *Account, evalAt time.Time) []string { } return nil } - -func clampProfitPreviewThreshold(threshold float64) float64 { - if math.IsNaN(threshold) || math.IsInf(threshold, 0) || threshold < 0 { - return 0 - } - return threshold -} - -// profitPreviewOverThreshold 与线上否决点共用同一 epsilon 边界语义。 -func profitPreviewOverThreshold(upstream, threshold float64) bool { - return upstream-threshold > profitControlRateEpsilon*math.Max(1, math.Abs(threshold)) -} diff --git a/backend/internal/service/scheduler_snapshot_service.go b/backend/internal/service/scheduler_snapshot_service.go index 38451b9b8e..542fc18b97 100644 --- a/backend/internal/service/scheduler_snapshot_service.go +++ b/backend/internal/service/scheduler_snapshot_service.go @@ -288,6 +288,17 @@ func (s *SchedulerSnapshotService) GetGroupByID(ctx context.Context, groupID int return s.groupRepo.GetByID(ctx, groupID) } +// GetGroupByIDLite 获取分组配置但不加载账号计数聚合。 +// 利润门只需要平台、倍率、利润与高峰字段，GetByID 附带的那条账号计数聚合 +// 查询纯属浪费——composite / 模型路由 / fallback 每次装门都要付一次，WS 更是 +// 每个 turn 一次，且发生在「是否启用利润控制」判定之前。 +func (s *SchedulerSnapshotService) GetGroupByIDLite(ctx context.Context, groupID int64) (*Group, error) { + if s.groupRepo == nil { + return nil, nil + } + return s.groupRepo.GetByIDLite(ctx, groupID) +} + // UpdateAccountInCache 立即更新 Redis 中单个账号的数据（用于模型限流后立即生效） func (s *SchedulerSnapshotService) UpdateAccountInCache(ctx context.Context, account *Account) error { if s.cache == nil || account == nil { diff --git a/frontend/src/i18n/locales/en/admin/overview.ts b/frontend/src/i18n/locales/en/admin/overview.ts index 462fa3cb27..c263f2caf8 100644 --- a/frontend/src/i18n/locales/en/admin/overview.ts +++ b/frontend/src/i18n/locales/en/admin/overview.ts @@ -1014,8 +1014,8 @@ export default { minMarginHint: 'Percent input, e.g. 30 means 30%; stored as a decimal on the backend', safetyBuffer: 'Safety buffer (%)', safetyBufferHint: 'Added to min margin and deducted from the downstream multiplier; defaults to 0', - marginRangeError: 'Min gross margin must be between 0 and 100 (exclusive)', - bufferRangeError: 'Safety buffer must be between 0 and 100 (exclusive)', + marginRangeError: 'Min gross margin must be between 0 and 99.99', + bufferRangeError: 'Safety buffer must be between 0 and 99.99', sumTooHigh: 'Min gross margin plus safety buffer must be less than 100%, otherwise every account would be excluded' }, modelsList: { diff --git a/frontend/src/i18n/locales/zh/admin/overview.ts b/frontend/src/i18n/locales/zh/admin/overview.ts index 3fa15e3a2e..e09f465b48 100644 --- a/frontend/src/i18n/locales/zh/admin/overview.ts +++ b/frontend/src/i18n/locales/zh/admin/overview.ts @@ -1011,8 +1011,8 @@ export default { minMarginHint: '百分比输入，如 30 表示 30%；后端按小数存储', safetyBuffer: '安全缓冲（%）', safetyBufferHint: '与最低毛利率相加后从下游倍率中扣除，默认 0', - marginRangeError: '最低毛利率应在 0 到 100 之间（不含 100）', - bufferRangeError: '安全缓冲应在 0 到 100 之间（不含 100）', + marginRangeError: '最低毛利率应在 0 到 99.99 之间', + bufferRangeError: '安全缓冲应在 0 到 99.99 之间', sumTooHigh: '最低毛利率与安全缓冲之和必须小于 100%，否则将排除全部账号' }, modelsList: { diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 25b41a6f59..cb0a698fb9 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -549,10 +549,6 @@ export interface Group { peak_start: string peak_end: string peak_rate_multiplier: number - // 分组利润控制（openai/anthropic/gemini/grok/antigravity 分组可启用；margin/buffer 为小数存储） - profit_control_enabled: boolean - profit_min_margin: number - profit_safety_buffer: number // Claude Code 客户端限制 claude_code_only: boolean fallback_group_id: number | null @@ -570,6 +566,12 @@ export interface Group { } export interface AdminGroup extends Group { + // 分组利润控制（openai/anthropic/gemini/grok/antigravity 分组可启用；margin/buffer 为小数存储）。 + // 仅管理员可见：与 rate_multiplier 相乘即可反推上游成本上限，不得下放到 Group。 + profit_control_enabled: boolean + profit_min_margin: number + profit_safety_buffer: number + // 模型路由配置（仅管理员可见，内部信息） model_routing: Record | null model_routing_enabled: boolean diff --git a/frontend/src/views/admin/__tests__/groupsProfitControl.spec.ts b/frontend/src/views/admin/__tests__/groupsProfitControl.spec.ts index 21e458281e..45a9262f5f 100644 --- a/frontend/src/views/admin/__tests__/groupsProfitControl.spec.ts +++ b/frontend/src/views/admin/__tests__/groupsProfitControl.spec.ts @@ -150,4 +150,32 @@ describe("validateProfitControlFormState", () => { ), ).toBe("sumTooHigh"); }); + + // 上界按换算后的小数判定：99.999% 会被 profitPercentToDecimal 四舍五入 + // 进位成 1.0，后端 validProfitControlRatio 要求严格 < 1，前端必须先拦下。 + it("rejects percents that round up to a decimal the backend rejects", () => { + expect(profitPercentToDecimal(99.999)).toBe(1); + expect( + validateProfitControlFormState( + formState({ profit_min_margin_percent: 99.999 }), + ), + ).toBe("marginRangeError"); + expect( + validateProfitControlFormState( + formState({ + profit_min_margin_percent: 0, + profit_safety_buffer_percent: 99.999, + }), + ), + ).toBe("bufferRangeError"); + }); + + it("still accepts the largest percent that survives the rounding", () => { + expect(profitPercentToDecimal(99.99)).toBe(0.9999); + expect( + validateProfitControlFormState( + formState({ profit_min_margin_percent: 99.99 }), + ), + ).toBeNull(); + }); }); diff --git a/frontend/src/views/admin/groupsProfitControl.ts b/frontend/src/views/admin/groupsProfitControl.ts index 6bb55761ed..71166e8208 100644 --- a/frontend/src/views/admin/groupsProfitControl.ts +++ b/frontend/src/views/admin/groupsProfitControl.ts @@ -32,24 +32,37 @@ export type ProfitControlFormState = { export const isProfitControlPlatform = (platform: string): boolean => ["openai", "anthropic", "gemini", "grok", "antigravity"].includes(platform); -// 提交前校验：margin/buffer 各自 ∈ [0,100)，且相加 < 100（否则阈值 <= 0， +// 提交前校验：margin/buffer 各自 ∈ [0,1)，且相加 < 1（否则阈值 <= 0， // 所有可核价账号都会被排除）。返回 null 表示通过，否则返回错误信息的 i18n key //（相对 admin.groups.profitControl 前缀）。仅支持平台且开关开启时才校验。 +// +// 上界校验必须落在 profitPercentToDecimal 的换算结果上，而不是界面百分比： +// 后端 validProfitControlRatio 校验的是小数 [0,1)，而 99.999% 会被四舍五入 +// 进位成 1.0——按百分比判 `< 100` 会让前端放行、后端 400。 export const validateProfitControlFormState = ( form: ProfitControlFormState, ): string | null => { if (!isProfitControlPlatform(form.platform) || !form.profit_control_enabled) { return null; } - const margin = Number(form.profit_min_margin_percent || 0); - const buffer = Number(form.profit_safety_buffer_percent || 0); - if (!Number.isFinite(margin) || margin < 0 || margin >= 100) { + const marginPercent = Number(form.profit_min_margin_percent || 0); + const bufferPercent = Number(form.profit_safety_buffer_percent || 0); + if (!Number.isFinite(marginPercent) || marginPercent < 0) { return "marginRangeError"; } - if (!Number.isFinite(buffer) || buffer < 0 || buffer >= 100) { + if (!Number.isFinite(bufferPercent) || bufferPercent < 0) { return "bufferRangeError"; } - if (margin + buffer >= 100) { + // 校验实际提交给后端的小数值，边界按构造对齐。 + const margin = profitPercentToDecimal(marginPercent); + const buffer = profitPercentToDecimal(bufferPercent); + if (margin >= 1) { + return "marginRangeError"; + } + if (buffer >= 1) { + return "bufferRangeError"; + } + if (margin + buffer >= 1) { return "sumTooHigh"; } return null;