fix(profit-control): decouple gate from image intent, propagate gate via selection, restore eager sticky fallback

Review fixes for the profit-control feature commit:

- Image intent no longer disables the profit gate. The shared /v1/responses
  handler previously skipped the pricing context (and therefore the gate)
  whenever the platform-wide image intent predicate matched, which includes
  Codex's passive image_gen namespace declaration: any client could disable
  admission control for anthropic/gemini/antigravity groups by declaring a
  namespace tool in the request body. Both /v1/responses paths now always
  install the token pricing context; image intent only drives capability
  routing and image billing. Mixed token+image requests stay token-gated;
  only dedicated media endpoints remain out of scope.
- Out-of-scope paths are now explicitly suppressed instead of implicitly
  ungated: Grok media (billed by media multipliers; also prevents in-flight
  video lookups from turning into spurious 404s), OpenAI-group count_tokens
  (unbilled), and Live calls (duration billed) carry a suppress marker that
  every install point honors, including the defensive scheduler-entry
  install.
- The gate resolved during selection now travels back to handlers on the
  AccountSelectionResult. The shared gateway installed the gate only on a
  scheduler-local context, so handler-side post-slot terminal rechecks and
  post-admission sticky binding were no-ops for anthropic/gemini/
  antigravity/shared-grok requests (and for composite-routed member groups
  on the OpenAI path). Handlers re-apply the carried gate via
  ContextWithSelectionProfitGate before the terminal recheck and binding;
  the WS acquired-selection branch gained the previously missing recheck.
- Sticky binding semantics restored for ungated traffic:
  BindStickySessionAfterProfitAdmission falls back to the official eager
  bind when no gate is installed (wait paths lost their only binding point
  otherwise), reads the pre-existing binding at bind time only when gated
  (removes the unconditional per-request Redis read the feature added to
  the shared handlers), and the legacy engine's three selection-time
  binding writes are skipped under a gate so a terminally vetoed account
  can no longer become the new sticky target.
- Responses WS connections re-freeze pricingAt and re-resolve the gate at
  every turn (BeforeTurn) and bill each turn with its own instant, closing
  the connect-at-valley/bill-at-valley window; a turn that fails the
  recheck closes the connection so the client reselects on reconnect.
- Terminal recheck no longer swaps a DB-fresh selected account for a
  staler snapshot object (UpdatedAt guard), and observer counters are
  documented as per-evaluation.
- Migrations renumbered 191/192 -> 192/193 after the passkey migration
  landed upstream as 191.

New regressions: selection-carried gate propagation (control group proves
the pre-fix no-op), image intent not disabling the shared gate, eager
binding fallback without a gate (both services), gated
read-failure/sentinel-miss binding semantics, legacy-engine deferred
binding under a gate with eager behavior preserved ungated, turn-level
pricing refresh (config re-resolution, scheduled-group precedence,
suppress, mid-connection disable), and suppress-marker coverage for the
request pricing context.
This commit is contained in:
Brisbanehuang authored and shaw committed 2026-08-01 22:39:32 +08:00
1 parent 20ad5ec506
commit fad2f215e8
22 files changed
+497 -95

No files matched your search

+18 -6
View File
@@ -424,7 +424,9 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
// Slot acquired: no longer waiting in queue.
releaseWait()
}
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(c.Request.Context(), account)
// 终检与准入后绑定使用选号结果携带的门(见 responses 同名注释)。
admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection)
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account)
if vetoed {
if accountReleaseFunc != nil {
accountReleaseFunc()
@@ -435,8 +437,12 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
}
account = latest
selection.Account = latest
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(c.Request.Context(), apiKey.GroupID, sessionKey, account.ID, sessionBoundAccountID); err != nil {
reqLog.Warn("gateway.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
// 等待路径保持既有 eager 绑定(无门时 helper 直接绑定);调度器已
// 抢槽的直达路径无门时由选号内部绑定,这里只在门下补准入后绑定。
if selection.ProfitGateActive() || !selection.Acquired {
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, apiKey.GroupID, sessionKey, account.ID); err != nil {
reqLog.Warn("gateway.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
}
}
// 账号槽位/等待计数需要在超时或断开时安全回收
accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc)
@@ -736,7 +742,9 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
// Slot acquired: no longer waiting in queue.
releaseWait()
}
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(c.Request.Context(), account)
// 终检与准入后绑定使用选号结果携带的门(见 responses 同名注释)。
admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection)
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account)
if vetoed {
if accountReleaseFunc != nil {
accountReleaseFunc()
@@ -747,8 +755,12 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
}
account = latest
selection.Account = latest
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(c.Request.Context(), currentAPIKey.GroupID, sessionKey, account.ID, sessionBoundAccountID); err != nil {
reqLog.Warn("gateway.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
// 等待路径保持既有 eager 绑定(无门时 helper 直接绑定);调度器已
// 抢槽的直达路径无门时由选号内部绑定,这里只在门下补准入后绑定。
if selection.ProfitGateActive() || !selection.Acquired {
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, currentAPIKey.GroupID, sessionKey, account.ID); err != nil {
reqLog.Warn("gateway.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
}
}
// 账号槽位/等待计数需要在超时或断开时安全回收
accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc)
@@ -159,8 +159,6 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) {
if groupPlatform == service.PlatformGemini && selectionSessionHash != "" {
selectionSessionHash = "gemini:" + selectionSessionHash
}
sessionBoundAccountID, _ := h.gatewayService.GetCachedSessionAccountID(c.Request.Context(), apiKey.GroupID, selectionSessionHash)
// 3. Account selection + failover loop
fs := NewFailoverState(h.maxAccountSwitches, false)
if groupPlatform == service.PlatformGemini {
@@ -226,7 +224,9 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) {
return
}
}
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(c.Request.Context(), account)
// 终检与准入后绑定使用选号结果携带的门(见 responses 同名注释)。
admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection)
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account)
if vetoed {
if accountReleaseFunc != nil {
accountReleaseFunc()
@@ -237,8 +237,10 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) {
}
account = latest
selection.Account = latest
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(c.Request.Context(), apiKey.GroupID, selectionSessionHash, account.ID, sessionBoundAccountID); err != nil {
reqLog.Warn("gateway.cc.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
if selection.ProfitGateActive() {
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, apiKey.GroupID, selectionSessionHash, account.ID); err != nil {
reqLog.Warn("gateway.cc.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
}
}
accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc)
@@ -90,11 +90,13 @@ func (h *GatewayHandler) Responses(c *gin.Context) {
setOpsRequestContext(c, reqModel, reqStream)
setOpsEndpointContext(c, "", int16(service.RequestTypeFromLegacy(reqStream, false)))
requestCtx := c.Request.Context()
pricingAt := time.Time{}
// 定价上下文无条件装配:/v1/responses 是 token 计费端点,声明生图工具的
// 混合请求同样按 token 计费(外加图片部分),其 token 利润保护不因请求体
// 里的任何工具声明(含 Codex 被动 image_gen namespace)而关闭。生图意图
// 仅用于能力路由与图片计费;独立图片/视频端点才在利润门范围之外。
requestCtx, pricingAt := service.WithGatewayTokenRequestPricing(requestCtx)
if service.IsImageGenerationIntentForPlatform("/v1/responses", reqModel, body, openAICompatibleRequestPlatform(c.Request.Context(), apiKey)) {
requestCtx = service.WithOpenAIImageGenerationIntent(requestCtx)
} else {
requestCtx, pricingAt = service.WithGatewayTokenRequestPricing(requestCtx)
}
c.Request = c.Request.WithContext(requestCtx)
@@ -161,7 +163,6 @@ func (h *GatewayHandler) Responses(c *gin.Context) {
APIKeyID: apiKey.ID,
}
sessionHash := h.gatewayService.GenerateSessionHash(parsedReq)
sessionBoundAccountID, _ := h.gatewayService.GetCachedSessionAccountID(requestCtx, apiKey.GroupID, sessionHash)
// 3. Account selection + failover loop
fs := NewFailoverState(h.maxAccountSwitches, false)
@@ -225,7 +226,11 @@ func (h *GatewayHandler) Responses(c *gin.Context) {
return
}
}
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(requestCtx, account)
// 终检与准入后绑定必须使用选号结果携带的门:门安装在调度栈的局部
// ctx 上(composite/fallback 还可能解析出与入口分组不同的门),直接用
// requestCtx 会退化为空操作。
admissionCtx := service.ContextWithSelectionProfitGate(requestCtx, selection)
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account)
if vetoed {
if accountReleaseFunc != nil {
accountReleaseFunc()
@@ -236,8 +241,10 @@ func (h *GatewayHandler) Responses(c *gin.Context) {
}
account = latest
selection.Account = latest
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(requestCtx, apiKey.GroupID, sessionHash, account.ID, sessionBoundAccountID); err != nil {
reqLog.Warn("gateway.responses.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
if selection.ProfitGateActive() {
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, apiKey.GroupID, sessionHash, account.ID); err != nil {
reqLog.Warn("gateway.responses.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
}
}
accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc)
@@ -355,7 +355,6 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
// 判断是否真的绑定了粘性会话:有 sessionKey 且已经绑定到某个账号
hasBoundSession := sessionKey != "" && sessionBoundAccountID > 0
profitStickyAccountID := sessionBoundAccountID
cleanedForUnknownBinding := false
fs := NewFailoverState(h.maxAccountSwitchesGemini, hasBoundSession)
@@ -470,7 +469,9 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
accountWaitCounted = false
}
}
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(c.Request.Context(), account)
// 终检与准入后绑定使用选号结果携带的门(见 responses 同名注释)。
admissionCtx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection)
latest, vetoed, reason := h.gatewayService.GatewayProfitControlVetoLatest(admissionCtx, account)
if vetoed {
if accountReleaseFunc != nil {
accountReleaseFunc()
@@ -481,8 +482,12 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
}
account = latest
selection.Account = latest
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(c.Request.Context(), apiKey.GroupID, sessionKey, account.ID, profitStickyAccountID); err != nil {
reqLog.Warn("gemini.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
// 等待路径保持既有 eager 绑定(无门时 helper 直接绑定);调度器已抢槽
// 的直达路径无门时由选号内部绑定,这里只在门下补准入后绑定。
if selection.ProfitGateActive() || !selection.Acquired {
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(admissionCtx, apiKey.GroupID, sessionKey, account.ID); err != nil {
reqLog.Warn("gemini.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
}
}
// 账号槽位/等待计数需要在超时或断开时安全回收
accountReleaseFunc = wrapReleaseOnDone(c.Request.Context(), accountReleaseFunc)
+5 -2
View File
@@ -177,7 +177,10 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.
return
}
}
requestCtx := c.Request.Context()
// Grok 媒体(图片/视频生成与视频查询)按媒体倍率计费,不在 token 利润门
// 范围内:显式豁免,防止 service 层防御性装门按文本 D 误过滤媒体请求,
// 也防止已计费的在途视频任务因绑定账号被门排除而查询返回伪 404。
requestCtx := service.WithOpenAIProfitControlSuppressed(c.Request.Context())
failedAccountIDs := make(map[int64]struct{})
sameAccountRetryCount := make(map[int64]int)
var lastFailoverErr *service.UpstreamFailoverError
@@ -295,7 +298,7 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.
accountReleaseFunc, slotResult := h.acquireResponsesAccountSlot(c, apiKey.GroupID, sessionHash, selection, false, &streamStarted, reqLog)
if slotResult == openAISlotAcquireProfitVetoed {
// Grok 分组非 openai 平台不装利润门,此分支实际不可达;防御性排除重选。
// 媒体路径已显式豁免利润门(suppress 标记),此分支仅防御性兜底。
failedAccountIDs[account.ID] = struct{}{}
continue
}
@@ -115,7 +115,7 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) {
// 分组利润控制:alpha search 文本入口请求级装门并固定 pricingAt
//(记录路径经 service.OpenAIPricingAtFromContext 从请求 ctx 回读)。
asPricingCtx, _ := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, false)
asPricingCtx, _ := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID)
c.Request = c.Request.WithContext(asPricingCtx)
for {
@@ -150,7 +150,7 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
var oauth429FailoverState service.OpenAIOAuth429FailoverState
// 分组利润控制:chat completions 文本入口请求级装门并固定 pricingAt。
ccPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, false)
ccPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID)
c.Request = c.Request.WithContext(ccPricingCtx)
for {
@@ -117,7 +117,7 @@ func (h *OpenAIGatewayHandler) Embeddings(c *gin.Context) {
routingStart := time.Now()
// 分组利润控制:embeddings 文本入口请求级装门并固定 pricingAt。
embPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, false)
embPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID)
c.Request = c.Request.WithContext(embPricingCtx)
for {
@@ -143,6 +143,9 @@ func (h *OpenAIGatewayHandler) CountTokens(c *gin.Context) {
}
requestStart := time.Now()
// count_tokens 不计费:显式豁免利润门,避免高倍率账号池被门排除后连
// token 计数都返回 no available accounts。
c.Request = c.Request.WithContext(service.WithOpenAIProfitControlSuppressed(c.Request.Context()))
sessionHash := h.gatewayService.GenerateSessionHash(c, body)
currentRoutingModel := routingModel
if preferredMappedModel != "" {
@@ -9,6 +9,7 @@ import (
"runtime/debug"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
@@ -440,9 +441,10 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
requiredCapability := openAIResponsesRequiredCapability(imageIntent, requestPlatform)
// 分组利润控制:请求级装配定价上下文——pricingAt 固定本请求的
// D 与计费高峰因子,选号、槽位终检与全部 failover 重入共用同一门与阈值;
// 显式生图意图跳门(图片/视频不在利润门范围)。
pricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, imageIntent)
// D 与计费高峰因子,选号、槽位终检与全部 failover 重入共用同一门与阈值。
// 生图意图只影响能力路由与图片计费,不关门:混合 /v1/responses 请求的
// token 计费部分仍受利润门保护,独立图片/视频端点才在门外。
pricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID)
c.Request = c.Request.WithContext(pricingCtx)
for {
@@ -1007,7 +1009,7 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
effectiveMappedModel := preferredMappedModel
// 分组利润控制:Messages 文本入口同样请求级装门并固定 pricingAt。
msgPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID, false)
msgPricingCtx, pricingAt := h.gatewayService.WithOpenAIRequestPricingContext(c.Request.Context(), apiKey.GroupID)
c.Request = c.Request.WithContext(msgPricingCtx)
for {
@@ -1398,7 +1400,9 @@ func (h *OpenAIGatewayHandler) acquireResponsesAccountSlot(
return nil, openAISlotAcquireFailed
}
ctx := c.Request.Context()
// 终检与准入后绑定使用选号结果携带的门:composite 等跨分组调度解析出的
// 门只存在于调度栈的局部 ctx,必须经选号结果重放到本函数的 ctx 上。
ctx := service.ContextWithSelectionProfitGate(c.Request.Context(), selection)
account := selection.Account
if selection.Acquired {
latest, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(ctx, account)
@@ -1411,6 +1415,13 @@ func (h *OpenAIGatewayHandler) acquireResponsesAccountSlot(
}
account = latest
selection.Account = latest
// 调度器已抢槽路径无门时由选号内部完成 eager 绑定;门下选号内部
// 推迟绑定,这里在终检通过后补准入后绑定。
if selection.ProfitGateActive() {
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(ctx, groupID, sessionHash, account.ID); err != nil {
reqLog.Warn("openai.bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
}
}
return wrapReleaseOnDone(ctx, selection.ReleaseFunc), openAISlotAcquireOK
}
if selection.WaitPlan == nil {
@@ -1786,10 +1797,11 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
requiredCapability = service.OpenAIEndpointCapabilityResponses
}
// 分组利润控制:WS 桥按连接装配定价上下文并装门(选号与抢槽
// 共用该 ctx);连接内不重选号,每个 turn 的计费共用同一 pricingAt——长
// 连接跨峰谷边界属已知残余风险,由安全缓冲承担。显式生图意图跳门。
wsPricingCtx, wsPricingAt := h.gatewayService.WithOpenAIRequestPricingContext(ctx, apiKey.GroupID, imageIntent)
// 分组利润控制:WS 桥按连接装配定价上下文并装门(选号与抢槽共用该
// ctx)。连接内不重选号,但每个 turn 开始经 BeforeTurn 重新冻结 pricingAt
// 并按最新门复核当前账号(准入与计费同源),峰前建连保活不能让后续 turn
// 继续按建连时刻的谷价计费。生图意图只影响能力路由与图片计费,不关门。
wsPricingCtx, wsPricingAt := h.gatewayService.WithOpenAIRequestPricingContext(ctx, apiKey.GroupID)
ctx = wsPricingCtx
for {
@@ -1837,7 +1849,24 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
if selection.WaitPlan != nil && selection.WaitPlan.MaxConcurrency > 0 {
accountMaxConcurrency = selection.WaitPlan.MaxConcurrency
}
// 终检、准入后绑定与后续 turn 级复核都使用选号结果携带的门(composite
// 等跨分组调度的门只存在于调度栈局部 ctx);准入成功后并入连接 ctx。
admissionCtx := service.ContextWithSelectionProfitGate(ctx, selection)
accountReleaseFunc := selection.ReleaseFunc
if selection.Acquired {
// 调度器已抢槽路径同样终检:选号与抢槽之间账号倍率可能刷新。
latest, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(admissionCtx, account)
if vetoed {
if accountReleaseFunc != nil {
accountReleaseFunc()
}
reqLog.Debug("openai.websocket_account_slot_profit_vetoed", zap.Int64("account_id", account.ID), zap.String("reason", reason))
failedAccountIDs[account.ID] = struct{}{}
continue
}
account = latest
selection.Account = latest
}
if !selection.Acquired {
if selection.WaitPlan == nil {
closeOpenAIClientWS(wsConn, coderws.StatusTryAgainLater, "account is busy, please retry later")
@@ -1859,7 +1888,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
}
// 分组利润控制:WS 快速抢槽成功后终检,越线则释放
// 槽位、排除该账号重新选号,全池耗尽由下一轮选号关闭连接。
latest, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(ctx, account)
latest, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(admissionCtx, account)
if vetoed {
if fastReleaseFunc != nil {
fastReleaseFunc()
@@ -1872,6 +1901,8 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
selection.Account = latest
accountReleaseFunc = fastReleaseFunc
}
// 准入完成:门并入连接 ctx,turn 级复核与 failover 重选共用。
ctx = admissionCtx
currentAccountRelease = wrapReleaseOnDone(ctx, accountReleaseFunc)
if err := h.gatewayService.BindStickySessionAfterProfitAdmission(ctx, apiKey.GroupID, sessionHash, account.ID); err != nil {
reqLog.Warn("openai.websocket_bind_sticky_session_after_profit_admission_failed", zap.Int64("account_id", account.ID), zap.Error(err))
@@ -1912,6 +1943,11 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
// turn-tagged slot preserves the exact mapping used for the in-flight request.
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
hooks := &service.OpenAIWSIngressHooks{
InitialRequestModel: reqModel,
MaxReasoningEffort: maxReasoningEffort,
@@ -1957,6 +1993,20 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
if cyberBlockedThisConn {
return service.NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, cyberSessionBlockedClientMsg, nil)
}
// 长连接跨峰谷/倍率刷新防护:每个 turn 按当前时刻重装门并复核
// 当前账号,越线即要求客户端重连重选(连接绑定单一上游账号,
// 无法中途换号)。本 turn 的准入与计费共用同一 pricingAt。
turnCtx, turnAt := h.gatewayService.WithOpenAITurnPricingContext(ctx, apiKey.GroupID)
if _, vetoed, reason := h.gatewayService.ProfitControlVetoLatest(turnCtx, account); vetoed {
reqLog.Info("openai.websocket_turn_profit_vetoed",
zap.Int("turn", turn),
zap.Int64("account_id", account.ID),
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()
if turn == 1 {
return nil
}
@@ -2055,6 +2105,9 @@ 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()
cyberBlocked := service.GetOpsCyberPolicy(c) != nil
h.submitOpenAIUsageRecordTask(ctx, result, func(taskCtx context.Context) {
if err := h.gatewayService.RecordUsage(taskCtx, &service.OpenAIRecordUsageInput{
@@ -2072,7 +2125,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
QuotaPlatform: quotaPlatform,
SessionID: sessionID,
ChannelUsageFields: turnUsageFields,
PricingAt: wsPricingAt,
PricingAt: turnRecordPricingAt,
CyberBlocked: cyberBlocked,
}); err != nil {
reqLog.Error("openai.websocket_record_usage_failed",
@@ -70,7 +70,10 @@ func profitSlotTestContext(t *testing.T, gw *service.OpenAIGatewayService, group
ProfitMinMargin: 0.5,
}
base := context.WithValue(context.Background(), ctxkey.Group, group)
ctx, pricingAt := gw.WithOpenAIRequestPricingContext(base, &groupID, suppress)
if suppress {
base = service.WithOpenAIProfitControlSuppressed(base)
}
ctx, pricingAt := gw.WithOpenAIRequestPricingContext(base, &groupID)
require.False(t, pricingAt.IsZero())
return ctx
}
@@ -94,7 +94,9 @@ func profitControlVetoLatest(ctx context.Context, selected *Account, snapshot *S
if err != nil || refreshed == nil {
slog.Warn("profit_control_account_refresh_failed", "group_id", gate.groupID, "platform", gate.platform, "account_id", selected.ID, "error", err)
openAIProfitControlObserverInstance.recordRefreshFailure(gate.groupID, gate.platform, gate.threshold)
} else {
} else if !refreshed.UpdatedAt.Before(selected.UpdatedAt) {
// 选号路径可能已做过 DB recheck,selected 比缓存快照更新鲜;只有
// 快照不落后时才替换,避免终检把新鲜账号换回较旧的缓存对象。
latest = refreshed
}
}
@@ -275,7 +275,6 @@ func TestGatewayProfitControlStickyVetoKeepsBindingUntilRateRecovers(t *testing.
&group.ID,
"sticky-profit",
cheap.ID,
expensive.ID,
))
require.Equal(t, expensive.ID, cache.sessionBindings["sticky-profit"], "终检通过的 fallback 账号也不得覆盖旧绑定")
require.Zero(t, cache.deletedSessions["sticky-profit"])
@@ -383,3 +382,91 @@ func TestGatewayProfitControlTerminalRefreshFailureFallsBackToSelectedObject(t *
require.False(t, vetoed)
require.Empty(t, reason)
}
// 选号结果携带门:门安装在调度栈局部 ctx 上,handler 必须经
// ContextWithSelectionProfitGate 重放后终检与准入后绑定才可见(评审修复回归)。
func TestGatewayProfitControlSelectionCarriesGateToHandlerContext(t *testing.T) {
group := gatewayProfitTestGroup(1, PlatformAnthropic)
svc := &GatewayService{}
expensive := gatewayProfitTestAccount(161, PlatformAnthropic, 0.9, group.ID)
gateCtx := svc.withGatewayProfitControlGate(gatewayProfitTestContext(group), &group.ID)
selection, err := svc.newSelectionResult(gateCtx, &expensive, true, nil, nil)
require.NoError(t, err)
require.True(t, selection.ProfitGateActive(), "选号结果必须携带调度栈内生效的门")
// 修复前的缺陷形态:handler 原始 ctx 不含门,终检退化为空操作。
_, vetoed, _ := svc.GatewayProfitControlVetoLatest(context.Background(), &expensive)
require.False(t, vetoed, "对照组:不重放门时终检确实看不到门")
handlerCtx := ContextWithSelectionProfitGate(context.Background(), selection)
latest, vetoed, reason := svc.GatewayProfitControlVetoLatest(handlerCtx, &expensive)
require.True(t, vetoed, "重放门后终检必须真实生效")
require.Equal(t, openAIProfitFilterReasonThreshold, reason)
require.NotNil(t, latest)
// 无门选号不携带门,重放为无操作。
plain, err := svc.newSelectionResult(context.Background(), &expensive, true, nil, nil)
require.NoError(t, err)
require.False(t, plain.ProfitGateActive())
require.Equal(t, context.Background(), ContextWithSelectionProfitGate(context.Background(), plain))
}
// 生图意图不关门(H1/H2 回归锚点):/v1/responses 混合请求即使带生图声明,
// token 定价上下文照常装配,共享门照常安装并否决越线账号。
func TestGatewayProfitControlImageIntentDoesNotDisableGate(t *testing.T) {
group := gatewayProfitTestGroup(2, PlatformAnthropic)
svc := &GatewayService{}
expensive := gatewayProfitTestAccount(162, PlatformAnthropic, 0.9, group.ID)
ctx := gatewayProfitTestContext(group)
ctx = WithOpenAIImageGenerationIntent(ctx)
gateCtx := svc.withGatewayProfitControlGate(ctx, &group.ID)
require.False(t, svc.isGatewayAccountProfitEligible(gateCtx, &expensive),
"请求体里的生图声明(含被动 image_gen namespace)不得关闭利润门")
}
// 无门时准入后绑定回退官方 eager 语义;门下读失败保守不写(评审 M-Bind 回归)。
func TestGatewayProfitControlAfterAdmissionBindSemantics(t *testing.T) {
groupID := int64(3)
expensiveID := int64(171)
cheapID := int64(172)
t.Run("eager without gate", func(t *testing.T) {
cache := &mockGatewayCacheForPlatform{sessionBindings: map[string]int64{"s": expensiveID}}
svc := &GatewayService{cache: cache}
require.NoError(t, svc.BindStickySessionAfterProfitAdmission(context.Background(), &groupID, "s", cheapID))
require.Equal(t, cheapID, cache.sessionBindings["s"], "无门时保持既有 eager 绑定行为")
})
t.Run("gated read failure is conservative", func(t *testing.T) {
// mock 的 miss 返回非 sentinel 错误,等价于 Redis 读失败:门下保守不写。
cache := &mockGatewayCacheForPlatform{sessionBindings: map[string]int64{}}
svc := &GatewayService{cache: cache}
gate := &openAIProfitControlGate{groupID: groupID, platform: PlatformAnthropic, threshold: 0.5}
gateCtx := context.WithValue(context.Background(), openAIProfitControlGateCtxKey{}, gate)
require.NoError(t, svc.BindStickySessionAfterProfitAdmission(gateCtx, &groupID, "absent", cheapID))
require.NotContains(t, cache.sessionBindings, "absent")
})
t.Run("gated sentinel miss binds", func(t *testing.T) {
cache := &sentinelMissGatewayCache{mockGatewayCacheForPlatform: &mockGatewayCacheForPlatform{sessionBindings: map[string]int64{}}}
svc := &GatewayService{cache: cache}
gate := &openAIProfitControlGate{groupID: groupID, platform: PlatformAnthropic, threshold: 0.5}
gateCtx := context.WithValue(context.Background(), openAIProfitControlGateCtxKey{}, gate)
require.NoError(t, svc.BindStickySessionAfterProfitAdmission(gateCtx, &groupID, "fresh", cheapID))
require.Equal(t, cheapID, cache.sessionBindings["fresh"], "门下无既有绑定(sentinel miss)应建立粘性")
})
}
// sentinelMissGatewayCache 让 miss 返回与真实仓库一致的 ErrStickySessionNotFound。
type sentinelMissGatewayCache struct {
*mockGatewayCacheForPlatform
}
func (c *sentinelMissGatewayCache) GetSessionAccountID(ctx context.Context, groupID int64, sessionHash string) (int64, error) {
if id, ok := c.sessionBindings[sessionHash]; ok {
return id, nil
}
return 0, ErrStickySessionNotFound
}
@@ -1453,12 +1453,12 @@ func (s *GatewayService) newSelectionResult(ctx context.Context, account *Accoun
if err != nil {
return nil, err
}
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: hydrated,
Acquired: acquired,
ReleaseFunc: release,
WaitPlan: waitPlan,
}, nil
}), nil
}
// filterByMinPriority 过滤出优先级最小的账号集合
+22 -5
View File
@@ -547,6 +547,15 @@ type AccountSelectionResult struct {
Acquired bool
ReleaseFunc func()
WaitPlan *AccountWaitPlan // nil means no wait allowed
// profitGate 携带本次选号真实生效的利润门(无门为 nil)。门安装在调度栈的
// 局部 ctx 上,handler 必须经 ContextWithSelectionProfitGate 重放后才能在
// 调度栈之外做抢槽后终检与准入后粘性绑定。
profitGate *openAIProfitControlGate
}
// ProfitGateActive 报告本次选号是否处于利润门之下。
func (r *AccountSelectionResult) ProfitGateActive() bool {
return r != nil && r.profitGate != nil
}
// ClaudeUsage 表示Claude API返回的usage信息
@@ -894,15 +903,23 @@ func (s *GatewayService) bindGatewayStickySessionDuringSelection(ctx context.Con
return s.BindStickySession(ctx, groupID, sessionHash, accountID)
}
// BindStickySessionAfterProfitAdmission records a successful terminal
// selection for a profit-controlled request without replacing the binding
// observed at request start. A temporarily ineligible sticky account remains
// bound and automatically becomes eligible again if its account rate recovers.
func (s *GatewayService) BindStickySessionAfterProfitAdmission(ctx context.Context, groupID *int64, sessionHash string, accountID, existingAccountID int64) error {
// BindStickySessionAfterProfitAdmission records a terminally admitted
// account. Without a profit gate it preserves the pre-existing eager binding
// behavior at the handler bind points. With a gate it never replaces a
// different binding that already exists: a temporarily ineligible sticky
// account remains bound and automatically becomes eligible again if its
// account rate recovers.
func (s *GatewayService) BindStickySessionAfterProfitAdmission(ctx context.Context, groupID *int64, sessionHash string, accountID int64) error {
if sessionHash == "" || accountID <= 0 || s.cache == nil {
return nil
}
if !gatewayProfitControlGateActive(ctx) {
return s.BindStickySession(ctx, groupID, sessionHash, accountID)
}
existingAccountID, err := s.cache.GetSessionAccountID(ctx, derefGroupID(groupID), sessionHash)
if err != nil && !errors.Is(err, ErrStickySessionNotFound) {
// 读失败时无法判断既有绑定,保守跳过而不是冒着覆盖健康绑定的风险写入。
slog.Warn("profit_control_sticky_binding_read_failed", "group_id", derefGroupID(groupID), "account_id", accountID, "error", err)
return nil
}
if existingAccountID > 0 && existingAccountID != accountID {
@@ -512,11 +512,11 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash(
result, acquireErr := s.service.tryAcquireAccountSlot(ctx, accountID, account.Concurrency)
if acquireErr == nil && result != nil && result.Acquired {
_ = s.service.refreshStickySessionTTL(ctx, req.GroupID, sessionHash, s.service.openAIWSSessionStickyTTL())
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: account,
Acquired: true,
ReleaseFunc: result.ReleaseFunc,
}, false, nil
}), false, nil
}
cfg := s.service.schedulingConfig()
@@ -532,7 +532,7 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash(
)
return nil, true, nil
}
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: account,
WaitPlan: &AccountWaitPlan{
AccountID: accountID,
@@ -540,7 +540,7 @@ func (s *defaultOpenAIAccountScheduler) selectBySessionHash(
Timeout: cfg.StickySessionWaitTimeout,
MaxWaiting: cfg.StickySessionMaxWaiting,
},
}, false, nil
}), false, nil
}
return nil, false, nil
}
@@ -1170,11 +1170,11 @@ func (s *defaultOpenAIAccountScheduler) tryAcquireOpenAISelectionOrderWithBudget
if req.SessionHash != "" && !req.PreserveStickyBinding {
_ = s.service.bindOpenAIStickySessionDuringSelection(ctx, req.GroupID, req.SessionHash, fresh.ID)
}
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: fresh,
Acquired: true,
ReleaseFunc: result.ReleaseFunc,
}, compactBlocked, nil
}), compactBlocked, nil
}
return nil, compactBlocked, nil
}
@@ -1249,15 +1249,15 @@ func (s *defaultOpenAIAccountScheduler) tryFallbackToWeightedSticky(
if req.SessionHash != "" && !req.PreserveStickyBinding {
_ = s.service.bindOpenAIStickySessionDuringSelection(ctx, req.GroupID, req.SessionHash, account.ID)
}
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: account,
Acquired: true,
ReleaseFunc: result.ReleaseFunc,
}, nil
}), nil
}
if s.service.concurrencyService != nil {
cfg := s.service.schedulingConfig()
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: account,
WaitPlan: &AccountWaitPlan{
AccountID: account.ID,
@@ -1265,7 +1265,7 @@ func (s *defaultOpenAIAccountScheduler) tryFallbackToWeightedSticky(
Timeout: cfg.StickySessionWaitTimeout,
MaxWaiting: cfg.StickySessionMaxWaiting,
},
}, nil
}), nil
}
}
return nil, nil
@@ -1622,7 +1622,7 @@ func (s *defaultOpenAIAccountScheduler) finishLoadBalanceSelectionFallback(
compactBlocked = true
continue
}
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: fresh,
WaitPlan: &AccountWaitPlan{
AccountID: fresh.ID,
@@ -1630,7 +1630,7 @@ func (s *defaultOpenAIAccountScheduler) finishLoadBalanceSelectionFallback(
Timeout: cfg.FallbackWaitTimeout,
MaxWaiting: cfg.FallbackMaxWaiting,
},
}, candidateCount, topK, loadSkew, nil
}), candidateCount, topK, loadSkew, nil
}
}
@@ -670,9 +670,10 @@ func (s *OpenAIGatewayService) selectAccountForModelWithExclusions(ctx context.C
return nil, err
}
// 4. 设置粘性会话绑定
// Set sticky session binding
if sessionHash != "" {
// 4. 设置粘性会话绑定(利润门下推迟到 handler 终检通过后再绑定,
// 终检否决的账号不得成为新的粘性目标;无门保持既有 eager 绑定与 TTL)
// Set sticky session binding (deferred until terminal admission under a profit gate)
if sessionHash != "" && !gatewayProfitControlGateActive(ctx) {
_ = s.setStickySessionAccountID(ctx, groupID, sessionHash, selected.ID, openaiStickySessionTTL)
}
@@ -1103,7 +1104,7 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex
if selectErr != nil {
return nil, true, selectErr
}
if sessionHash != "" {
if sessionHash != "" && !gatewayProfitControlGateActive(ctx) {
_ = s.setStickySessionAccountID(ctx, groupID, sessionHash, fresh.ID, openaiStickySessionTTL)
}
return selection, true, nil
@@ -1142,7 +1143,7 @@ func (s *OpenAIGatewayService) selectAccountWithLoadAwareness(ctx context.Contex
if selectErr != nil {
return nil, selectErr
}
if sessionHash != "" {
if sessionHash != "" && !gatewayProfitControlGateActive(ctx) {
_ = s.setStickySessionAccountID(ctx, groupID, sessionHash, fresh.ID, openaiStickySessionTTL)
}
return selection, nil
@@ -1354,12 +1355,12 @@ func (s *OpenAIGatewayService) newSelectionResult(ctx context.Context, account *
if err != nil {
return nil, err
}
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: hydrated,
Acquired: acquired,
ReleaseFunc: release,
WaitPlan: waitPlan,
}, nil
}), nil
}
func (s *OpenAIGatewayService) newAcquiredSelectionResult(ctx context.Context, account *Account, release func()) (*AccountSelectionResult, error) {
+3
View File
@@ -144,6 +144,9 @@ func (s *OpenAIGatewayService) CreateLiveCall(
}
excluded := make(map[int64]struct{})
// Live 按通话时长计费,不属于 token 利润门的语义范围:显式豁免,避免
// 防御性装门按文本 D 过滤 Live 账号池且门与计费时刻不同源。
ctx = WithOpenAIProfitControlSuppressed(ctx)
var lastErr error
for attempt := 0; attempt <= 3; attempt++ {
selection, _, selectErr := s.SelectAccountWithSchedulerForCapability(
@@ -22,14 +22,23 @@ package service
//
// 装门点(gate 随 ctx 传播,请求内复用,覆盖等待/重试/failover/抢槽后终检):
// - handler 各文本入口经 WithOpenAIRequestPricingContext 在请求开始统一装门并
// 固定 pricingAt;显式生图意图(responses image_generation)以抑制标记跳门,
// 图片/视频保持既有调度不装门。
// 固定 pricingAt。生图意图只用于能力路由与图片计费,不决定是否装门:混合
// /v1/responses 请求(含声明了生图工具的请求)的 token 计费部分仍受利润门
// 保护,请求体内的任何工具声明都不能作为关门开关。
// - 门范围之外的路径显式携带 WithOpenAIProfitControlSuppressed 标记:独立
// 图片/视频端点与 Grok 媒体(媒体计费另有倍率来源)、count_tokens(不计费)、
// live(按时长计费)。所有装门点(含防御性装门)都尊重该标记。
// - selectAccountWithScheduler 顶部:唯一文本调度入口的防御性装门(ctx 已有
// 同分组门则复用,保证 failover 阈值稳定);requiredImageCapability != "" 或
// requiredCapability == OpenAIEndpointCapabilityResponses(该值当前仅显式
// 生图意图设置)不装门。
// 同分组门则复用,保证 failover 阈值稳定);requiredImageCapability != ""
// 的专用图片调度不装门。
// - 公开 SelectAccountWithLoadAwareness / SelectAccountByPreviousResponseID:
// 防御性装门,保证不经唯一入口的调用方无法绕过。
// - 选号结果通过 AccountSelectionResult 携带真实生效的门(composite/fallback
// 路由可能解析出与入口分组不同的门),handler 经
// ContextWithSelectionProfitGate 重放后再做抢槽后终检与准入后粘性绑定。
// - 长连接(Responses WS)在每个 turn 开始经 WithOpenAITurnPricingContext
// 重新冻结 pricingAt 并重装门:turn 的准入与计费同源,跨峰谷边界不再共用
// 建连时刻的 D。
//
// 否决点(消费 gate,任何 fallback 都无法把已排除账号重新放回):
// - defaultOpenAIAccountScheduler.isAccountRequestCompatibleReason:候选池
@@ -78,9 +87,9 @@ const (
type openAIProfitControlGateCtxKey struct{}
// openAIProfitControlSuppressCtxKey 标记本请求显式跳过利润门(生图意图等
// 利润门范围外流量)。所有装门点看到该标记后一律不装门,防止 service 层防御性
// 装门把边界外流量重新拉回利润过滤。
// openAIProfitControlSuppressCtxKey 标记本请求显式跳过利润门(独立图片/视频
// 端点、Grok 媒体、count_tokens、live 等利润门范围外流量)。所有装门点看到该
// 标记后一律不装门,防止 service 层防御性装门把边界外流量重新拉回利润过滤。
type openAIProfitControlSuppressCtxKey struct{}
// openAIPricingAtCtxKey 携带请求级定价时刻 pricingAt:门的 D 与 RecordUsage
@@ -103,17 +112,49 @@ type openAIProfitControlGate struct {
// WithOpenAIRequestPricingContext 在请求开始处装配请求级定价上下文:固定
// pricingAt(返回给调用方,供 RecordUsage 入参共用同一时刻),并按分组安装
// 利润门;suppressProfitGate 为 true(显式生图意图)时只固定 pricingAt、写入
// 抑制标记,不装门。handler 各文本入口应在选号循环前调用一次。
func (s *OpenAIGatewayService) WithOpenAIRequestPricingContext(ctx context.Context, groupID *int64, suppressProfitGate bool) (context.Context, time.Time) {
// 利润门。ctx 携带 WithOpenAIProfitControlSuppressed 标记(门范围外流量)时
// 只固定 pricingAt、不装门。handler 各文本入口应在选号循环前调用一次。
func (s *OpenAIGatewayService) WithOpenAIRequestPricingContext(ctx context.Context, groupID *int64) (context.Context, time.Time) {
pricingAt := timezone.Now()
ctx = context.WithValue(ctx, openAIPricingAtCtxKey{}, pricingAt)
if suppressProfitGate {
return context.WithValue(ctx, openAIProfitControlSuppressCtxKey{}, struct{}{}), pricingAt
}
return s.withOpenAIProfitControlGate(ctx, groupID), pricingAt
}
// WithOpenAIProfitControlSuppressed 标记本请求在利润门范围之外(独立图片/视频
// 端点、Grok 媒体、count_tokens、live)。所有装门点(含 service 层防御性装门)
// 都尊重该标记;它只关闭利润准入过滤,不影响定价上下文与计费。
func WithOpenAIProfitControlSuppressed(ctx context.Context) context.Context {
return context.WithValue(ctx, openAIProfitControlSuppressCtxKey{}, struct{}{})
}
// WithOpenAITurnPricingContext 在长连接(Responses WS)的每个 turn 开始重新
// 冻结 pricingAt 并按当前配置重装利润门,使 turn 的准入与计费同源:峰前建连
// 保活不再让后续 turn 继续按建连时刻的谷价定价。连接可能被调度到与入口分组
// 不同的分组(composite 成员分组),turn 级重装以连接上已装门的调度分组为准;
// 连接从未装门时才回退入口分组。抑制标记下只刷新 pricingAt。
func (s *OpenAIGatewayService) WithOpenAITurnPricingContext(ctx context.Context, groupID *int64) (context.Context, time.Time) {
pricingAt := timezone.Now()
ctx = context.WithValue(ctx, openAIPricingAtCtxKey{}, pricingAt)
if _, suppressed := ctx.Value(openAIProfitControlSuppressCtxKey{}).(struct{}); suppressed {
return ctx, pricingAt
}
if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing != nil {
gid := existing.groupID
groupID = &gid
}
gate := s.resolveOpenAIProfitControlGate(ctx, groupID)
if gate == nil {
// 分组已关门(或配置读取失败 fail-open):清除旧 turn 的门,后续 turn
// 按无门放行,与 HTTP 路径的开关语义一致。
if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing != nil {
return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, (*openAIProfitControlGate)(nil)), pricingAt
}
return ctx, pricingAt
}
openAIProfitControlObserverInstance.recordInstall(gate.groupID, gate.platform, gate.threshold)
return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, gate), pricingAt
}
// openAIPricingAtFromContext 返回请求级定价时刻;未装配(内部调用、非文本
// 入口)时 ok=false,调用方回退 timezone.Now() 保持既有行为。
func openAIPricingAtFromContext(ctx context.Context) (time.Time, bool) {
@@ -215,6 +256,34 @@ func (s *OpenAIGatewayService) resolveOpenAIProfitControlGate(ctx context.Contex
}
}
// attachSelectionProfitGate 把调度上下文里生效的利润门记录到选号结果上。门在
// 选号函数内部的局部 ctx 上安装(composite/fallback 路由还可能解析出与入口
// 分组不同的门),不随返回值离开调度栈;结果携带后 handler 才能对"真实过滤了
// 候选的那个门"做抢槽后终检与准入后绑定。
func attachSelectionProfitGate(ctx context.Context, sel *AccountSelectionResult) *AccountSelectionResult {
if sel == nil {
return nil
}
if gate, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && gate != nil {
sel.profitGate = gate
}
return sel
}
// ContextWithSelectionProfitGate 把选号时真实生效的利润门重放到 ctx 上。
// handler 在拿到选号结果后必须用返回的 ctx 做抢槽后终检
// (ProfitControlVetoLatest / GatewayProfitControlVetoLatest)与准入后粘性
// 绑定,否则这两步会因为看不到调度栈内安装的门而退化为空操作。
func ContextWithSelectionProfitGate(ctx context.Context, sel *AccountSelectionResult) context.Context {
if sel == nil || sel.profitGate == nil {
return ctx
}
if existing, ok := ctx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate); ok && existing == sel.profitGate {
return ctx
}
return context.WithValue(ctx, openAIProfitControlGateCtxKey{}, sel.profitGate)
}
// openAIProfitControlVetoReason 报告利润门是否否决该账号。ctx 中没有门
// (分组未启用利润控制或本请求跳门)或账号为 nil 时一律放行。
func openAIProfitControlVetoReason(ctx context.Context, account *Account) (bool, string) {
@@ -267,13 +336,18 @@ func (s *OpenAIGatewayService) bindOpenAIStickySessionDuringSelection(ctx contex
}
// BindStickySessionAfterProfitAdmission records the terminally admitted
// account without overwriting a different binding that already exists. A
// temporarily ineligible account therefore remains sticky and becomes
// eligible again automatically after its rate recovers.
// account. Without a profit gate it preserves the pre-existing eager binding
// behavior at the handler bind points. With a gate it never overwrites a
// different binding that already exists, so a temporarily ineligible account
// remains sticky and becomes eligible again automatically after its rate
// recovers.
func (s *OpenAIGatewayService) BindStickySessionAfterProfitAdmission(ctx context.Context, groupID *int64, sessionHash string, accountID int64) error {
if sessionHash == "" || accountID <= 0 || !gatewayProfitControlGateActive(ctx) {
if sessionHash == "" || accountID <= 0 {
return nil
}
if !gatewayProfitControlGateActive(ctx) {
return s.BindStickySession(ctx, groupID, sessionHash, accountID)
}
existingAccountID, err := s.getStickySessionAccountID(ctx, groupID, sessionHash)
if err != nil && !errors.Is(err, ErrStickySessionNotFound) {
slog.Warn("profit_control_sticky_binding_read_failed", "group_id", derefGroupID(groupID), "account_id", accountID, "error", err)
@@ -286,6 +360,10 @@ func (s *OpenAIGatewayService) BindStickySessionAfterProfitAdmission(ctx context
}
// ---- 可观测性:按分组累计计数 + 采样日志(无逐请求输出) ----
//
// 计数按"每次准入评估"累计,而非每请求:粘性层校验与候选池过滤可能对同一账号
// 各评估一次,failover 重入也会再次计数。计数用于确认门在真实流量上生效及否决
// 构成,不能当作精确的请求数或账号数。
type openAIProfitControlGroupStats struct {
installs atomic.Int64
@@ -314,3 +314,57 @@ func TestProfitControl_EligibilityFunctionVetoes(t *testing.T) {
// 无门时保持既有行为。
require.True(t, isOpenAICompatibleAccountEligibleForRequest(context.Background(), expensive, PlatformOpenAI, "", false, ""))
}
// legacy 引擎粘性写回(评审 M-Legacy 回归):门下选号阶段不得直写粘性——
// 终检否决的账号不能成为新绑定;无门保持官方 eager 绑定与原 TTL 语义。
func TestProfitControl_LegacyEngineDefersStickyBindingUnderGate(t *testing.T) {
now := time.Now()
cheap := upstreamCostTestAccount(45, UpstreamBillingProbeStatusOK, 0.3, now.Add(-time.Minute), 30*time.Minute)
expensive := upstreamCostTestAccount(46, UpstreamBillingProbeStatusOK, 0.8, now.Add(-time.Minute), 30*time.Minute)
profitControlTestAccountWithRate(cheap, 0.3)
profitControlTestAccountWithRate(expensive, 0.8)
for _, account := range []*Account{cheap, expensive} {
account.Status = StatusActive
account.Schedulable = true
account.Concurrency = 2
}
groupID := int64(9)
const sessionHash = "legacy-sticky"
newSvc := func(bindings map[string]int64) (*OpenAIGatewayService, *schedulerTestGatewayCache) {
cache := &schedulerTestGatewayCache{sessionBindings: bindings}
return &OpenAIGatewayService{
accountRepo: stubOpenAIAccountRepo{accounts: []Account{*cheap, *expensive}},
cfg: &config.Config{},
rateLimitService: newOpenAIAdvancedSchedulerRateLimitService("false"),
concurrencyService: NewConcurrencyService(stubConcurrencyCache{}),
cache: cache,
}, cache
}
t.Run("gated selection defers binding to terminal admission", func(t *testing.T) {
svc, cache := newSvc(map[string]int64{})
ctx := profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0))
selection, _, err := svc.SelectAccountWithScheduler(ctx, &groupID, "", sessionHash, "gpt-test", nil, OpenAIUpstreamTransportAny, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.Equal(t, cheap.ID, selection.Account.ID)
require.Empty(t, cache.sessionBindings, "门下 legacy 选号阶段不得直写粘性绑定")
require.True(t, selection.ProfitGateActive(), "legacy 选号结果同样携带门")
if selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
})
t.Run("ungated selection keeps official eager binding", func(t *testing.T) {
svc, cache := newSvc(map[string]int64{})
group := profitControlTestGroup(groupID, 0.5, 0)
group.ProfitControlEnabled = false
selection, _, err := svc.SelectAccountWithScheduler(profitControlTestCtx(group), &groupID, "", sessionHash, "gpt-test", nil, OpenAIUpstreamTransportAny, false)
require.NoError(t, err)
require.NotNil(t, selection)
require.NotEmpty(t, cache.sessionBindings, "无门时 legacy 选号保持官方 eager 绑定")
if selection.ReleaseFunc != nil {
selection.ReleaseFunc()
}
})
}
@@ -14,8 +14,8 @@ import (
"github.com/stretchr/testify/require"
)
// WithOpenAIRequestPricingContext:装门 + 固定 pricingAt;抑制标记跳门且防御性
// 装门无法把门加回来。
// WithOpenAIRequestPricingContext:装门 + 固定 pricingAt;显式抑制标记
// (媒体/count_tokens/live 等门范围外路径)跳门且防御性装门无法把门加回来。
func TestProfitControl_RequestPricingContext(t *testing.T) {
svc := &OpenAIGatewayService{}
groupID := int64(61)
@@ -25,7 +25,7 @@ func TestProfitControl_RequestPricingContext(t *testing.T) {
t.Run("installs gate and pricing instant", func(t *testing.T) {
base := profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0))
ctx, pricingAt := svc.WithOpenAIRequestPricingContext(base, &groupID, false)
ctx, pricingAt := svc.WithOpenAIRequestPricingContext(base, &groupID)
require.False(t, pricingAt.IsZero())
require.Equal(t, pricingAt, OpenAIPricingAtFromContext(ctx))
vetoed, reason := OpenAIProfitControlVeto(ctx, expensive)
@@ -33,9 +33,9 @@ func TestProfitControl_RequestPricingContext(t *testing.T) {
require.Equal(t, openAIProfitFilterReasonThreshold, reason)
})
t.Run("image intent suppresses gate everywhere", func(t *testing.T) {
base := profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0))
ctx, pricingAt := svc.WithOpenAIRequestPricingContext(base, &groupID, true)
t.Run("suppress marker skips gate everywhere", func(t *testing.T) {
base := WithOpenAIProfitControlSuppressed(profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0)))
ctx, pricingAt := svc.WithOpenAIRequestPricingContext(base, &groupID)
require.False(t, pricingAt.IsZero(), "跳门时 pricingAt 仍需固定供计费共用")
vetoed, _ := OpenAIProfitControlVeto(ctx, expensive)
require.False(t, vetoed)
@@ -199,3 +199,75 @@ func TestOpenAIProfitControlStickyBindingOccursOnlyAfterTerminalAdmission(t *tes
require.NoError(t, svc.BindStickySessionAfterProfitAdmission(ctx, &groupID, sessionHash, cheapID))
require.Equal(t, cheapID, cache.sessionBindings[cacheKey], "无既有绑定时应在终检通过后建立粘性")
}
// WithOpenAITurnPricingContext:长连接 turn 边界重新冻结 pricingAt 并按当前
// 配置重装门(区别于请求级同门复用);已装门时以门所属调度分组为准。
func TestProfitControl_TurnPricingContext(t *testing.T) {
svc := &OpenAIGatewayService{}
groupID := int64(63)
expensive := upstreamCostTestAccount(3, UpstreamBillingProbeStatusOK, 0.8, time.Now().Add(-time.Minute), 30*time.Minute)
profitControlTestAccountWithRate(expensive, 0.8)
t.Run("refreshes instant and re-resolves gate config", func(t *testing.T) {
group := profitControlTestGroup(groupID, 0.5, 0)
base := profitControlTestCtx(group)
connCtx, connAt := svc.WithOpenAIRequestPricingContext(base, &groupID)
vetoed, _ := OpenAIProfitControlVeto(connCtx, expensive)
require.True(t, vetoed)
// 连接中途运营者放宽 margin:turn 级重装必须生效(请求级复用不生效)。
group.ProfitMinMargin = 0.1
turnCtx, turnAt := svc.WithOpenAITurnPricingContext(connCtx, &groupID)
require.False(t, turnAt.Before(connAt))
require.Equal(t, turnAt, OpenAIPricingAtFromContext(turnCtx))
vetoed, _ = OpenAIProfitControlVeto(turnCtx, expensive)
require.False(t, vetoed, "turn 级重装应采用最新分组配置")
})
t.Run("keeps scheduled group of the existing gate", func(t *testing.T) {
scheduledGroupID := int64(64)
scheduled := profitControlTestGroup(scheduledGroupID, 0.5, 0)
connCtx, _ := svc.WithOpenAIRequestPricingContext(profitControlTestCtx(scheduled), &scheduledGroupID)
// 入口分组与调度分组不同(composite 成员分组场景):turn 重装取门的分组。
entryGroupID := int64(65)
turnCtx, _ := svc.WithOpenAITurnPricingContext(connCtx, &entryGroupID)
gate, ok := turnCtx.Value(openAIProfitControlGateCtxKey{}).(*openAIProfitControlGate)
require.True(t, ok)
require.NotNil(t, gate)
require.Equal(t, scheduledGroupID, gate.groupID)
})
t.Run("suppress marker only refreshes instant", func(t *testing.T) {
base := WithOpenAIProfitControlSuppressed(profitControlTestCtx(profitControlTestGroup(groupID, 0.5, 0)))
turnCtx, turnAt := svc.WithOpenAITurnPricingContext(base, &groupID)
require.False(t, turnAt.IsZero())
vetoed, _ := OpenAIProfitControlVeto(turnCtx, expensive)
require.False(t, vetoed)
})
t.Run("clears gate when group disables profit control mid-connection", func(t *testing.T) {
group := profitControlTestGroup(groupID, 0.5, 0)
connCtx, _ := svc.WithOpenAIRequestPricingContext(profitControlTestCtx(group), &groupID)
group.ProfitControlEnabled = false
turnCtx, _ := svc.WithOpenAITurnPricingContext(connCtx, &groupID)
vetoed, _ := OpenAIProfitControlVeto(turnCtx, expensive)
require.False(t, vetoed, "关门后 turn 级复核应放行")
})
}
// 无门时准入后绑定回退官方 eager 语义:等待/抢槽路径不得因利润控制关闭而
// 失去粘性绑定(评审 M-Bind 回归锚点)。
func TestOpenAIProfitControlAfterAdmissionBindEagerWithoutGate(t *testing.T) {
groupID := int64(82)
expensiveID := int64(903)
cheapID := int64(904)
const sessionHash = "no-gate-sticky"
const cacheKey = "openai:" + sessionHash
cache := &schedulerTestGatewayCache{
sessionBindings: map[string]int64{cacheKey: expensiveID},
}
svc := &OpenAIGatewayService{cache: cache}
require.NoError(t, svc.BindStickySessionAfterProfitAdmission(context.Background(), &groupID, sessionHash, cheapID))
require.Equal(t, cheapID, cache.sessionBindings[cacheKey], "无门时保持既有 eager 绑定行为")
}
@@ -415,16 +415,16 @@ func (s *OpenAIGatewayService) selectAccountByPreviousResponseIDForCapability(
responseID,
store.BindResponseAccount(ctx, derefGroupID(groupID), responseID, accountID, s.openAIWSResponseStickyTTL()),
)
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: account,
Acquired: true,
ReleaseFunc: result.ReleaseFunc,
}, nil
}), nil
}
cfg := s.schedulingConfig()
if s.concurrencyService != nil {
return &AccountSelectionResult{
return attachSelectionProfitGate(ctx, &AccountSelectionResult{
Account: account,
WaitPlan: &AccountWaitPlan{
AccountID: accountID,
@@ -432,7 +432,7 @@ func (s *OpenAIGatewayService) selectAccountByPreviousResponseIDForCapability(
Timeout: cfg.StickySessionWaitTimeout,
MaxWaiting: cfg.StickySessionMaxWaiting,
},
}, nil
}), nil
}
return nil, nil
}