fix(profit-control): stop leaking profit policy, close veto livelock, restore passthrough turn pricing

审计修复,逐条如下。

H1 利润策略泄露给所有普通用户
  profit_control_enabled / profit_min_margin / profit_safety_buffer 从
  dto.Group 移到 dto.AdminGroup(后者内嵌前者),赋值相应从
  groupFromServiceBase 移到 GroupFromServiceAdmin;前端 TS 同步从 Group 移到
  AdminGroup。dto.Group 是 GET /api/v1/groups/available 的响应体,该响应本就带
  rate_multiplier,相乘即可反推运营方上游采购成本上限。
  api_contract_test.go 的 /groups/available golden JSON 回滚这三个字段,并把
  fixture 改成非零值(require.JSONEq 是精确比对,缺字段即失败)。
  新增 dto 层边界测试:普通用户 DTO 不含三字段、管理员 DTO 仍含。

M1 利润终检 continue 与 failover 503 退避互动产生活锁
  FailoverState 新增 profitVetoedAccountIDs / profitVetoCount 与
  RecordProfitVeto():加入排除集 + 计数,达 maxProfitVetoAttempts(10) 返回
  FailoverExhausted。HandleSelectionExhausted 的 503 清空分支改为清空后把利润
  否决的账号放回排除集;若排除集已全部由利润否决贡献,清空不会带来任何新候选,
  直接判定耗尽(否则 SwitchCount 永不前进、退避条件永远成立,每 2s 空转一轮)。
  五个 handler 否决点(gateway_handler ×2 / responses / chat_completions /
  gemini_v1beta)改为经 RecordProfitVeto 决策,耗尽时按无可用账号终止。
  回归测试钉死:503 之后持续利润否决必须有限步终止且不 spin;未启用利润控制的
  请求退避语义完全不变。

M2 排队等槽后才终检,延迟可放大到 N × WaitPlan.Timeout
  OpenAI 侧选号循环(自有 failedAccountIDs map,非 FailoverState)新增
  recordOpenAIProfitVeto + handleOpenAIProfitVetoExhausted,共用同一上限语义。
  覆盖 responses / messages-dispatch / chat_completions / alpha_search /
  embeddings / images / grok_media 七处,以及 WS 两处否决分支。

M4 ws_v2 透传 ingress 绕过 per-turn 重定价(选方案 B:最小止血)
  透传 relay 只回调 AfterTurn、没有任何 turn 起始回调,hooks.BeforeTurn 永远
  不触发,而 handler 把 turnPricingAt 初始化成建连时刻 ⇒ 透传连接全部 turn 按
  建连时刻的高峰因子结算,客户端峰前建连保活即可全程谷价——正是本 PR 想堵的
  漏洞。改为 openAIWSTurnPricing 零值起步、只由 BeforeTurn 冻结;透传路径保持
  零值,RecordUsage 回退记录时刻,与引入利润控制前的基线一致。
  未选方案 A(给透传补 turn 起始回调):passthrough_relay.go 是 #5167 刚修过的
  取消传播/close frame 时序敏感区;且 BeforeTurn 还承担 turn>1 的并发槽位抢占,
  接进去等于给透传连接引入 per-turn 抢槽,风险远超本次修复范围。透传仍有建连时
  的准入门,只是没有 turn 级复核,已在两处注释写明。
  测试:service 层钉死透传 ingress 不触发 BeforeTurn(含失败时的复核指引),
  handler 层钉死零值语义与逐 turn 覆盖。

M5 装门读分组走了带账号计数聚合的 GetByID
  SchedulerSnapshotService 新增 GetGroupByIDLite,openai/gateway 两处装门改用
  之。门只需要平台/倍率/利润/高峰字段,且该查询发生在「是否启用利润控制」判定
  之前,未启用的分组同样付代价。两个测试 stub 的 GetByID 改成 panic 守卫。

M6 认证快照注释与真实读取路径相反
  门解析优先取 ctxkey.Group,而它就是本快照物化出来的对象,直连流量走的正是这
  条路。改正注释,与 api_key_repo.go 投影处的说明对齐,避免后人照旧注释删列。

M3 rate_multiplier 为 nil 时利润门 fail-closed(不改行为,加护栏)
  保留 fail-closed。补 repository 层测试钉死账号调度快照的 full/metadata 两份
  payload 都必须保留 RateMultiplier(含 0 值),漏列在 CI 就红。

L1 迁移号注释 191 / 191-192 改为实际的 192/193。
L2 admin group Create 的利润配置预校验改用与 CreateGroup 一致的归一化平台
   (新增 service.NormalizeGroupPlatform,两边共用)。保留预校验而非删除:
   service 层返回的是无类型 error,经 ErrorFrom 会变成 500,删掉会把合法的
   400 降级成 500。
L3 前端利润校验的上界改为判定换算后的小数(后端按小数校验 [0,1)),
   99.999% 会四舍五入进位成 1.0 而被后端 400;i18n en/zh 同步改为 0-99.99。
L4 clampProfitControlThreshold / profitControlOverThreshold 抽为共用函数,
   线上装门/否决点与 profit-preview 不再各自实现,附边界语义测试。
L5 profit-preview 补「默认 D 有账号但最低有效 D 归零」的告警(两档都为 0 由
   既有告警覆盖,不重复)。
This commit is contained in:
shaw
2026-08-01 22:39:33 +08:00
parent fad2f215e8
commit dec47e8fae
41 changed files with 1041 additions and 125 deletions
+18
View File
@@ -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
}
+36
View File
@@ -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)
}
}
}
+1 -1
View File
@@ -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).
@@ -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
}
@@ -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)
}
}
}
+3 -3
View File
@@ -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,
+17 -14
View File
@@ -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"`
+76 -4
View File
@@ -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
@@ -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, "无利润否决时排除列表应被完全清空")
}
+12 -2
View File
@@ -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
@@ -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
@@ -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
@@ -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
+7 -2
View File
@@ -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 {
@@ -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 {
@@ -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 {
@@ -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 {
@@ -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{
+6 -2
View File
@@ -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 {
@@ -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, "两条路径的利润否决上限必须一致")
}
@@ -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 必须使用自己的定价时刻")
}
@@ -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)
})
}
+15 -14
View File
@@ -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",
+1 -4
View File
@@ -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)
@@ -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"`
@@ -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)
}
@@ -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)
+11
View File
@@ -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,所有可核价账号都会被排除,视为配置错误而不是静默全黑)。
@@ -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)
}
}
@@ -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
}
@@ -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)
@@ -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,
@@ -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 提交用量")
}
@@ -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))
}
+4 -16
View File
@@ -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))
}
@@ -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 {
@@ -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: {
@@ -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: {
+6 -4
View File
@@ -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<string, number[]> | null
model_routing_enabled: boolean
@@ -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();
});
});
@@ -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;