feat(quota): sentinel 回填消除无配额行用户 preflight 每请求回源 DB
无 user×platform 配额行的用户,preflight 每次 cache MISS 后回源 DB 查得"无行" 却不缓存该结论,导致每请求一次 DB 往返。本 PR 回填 sentinel 占位 entry,使后续 请求命中 Redis 后稳定判"无 limit",TTL 内不再查 DB。 - config: 加 UserPlatformQuotaSentinelTTLSeconds(默认 3600s,短于普通 quota cache 的 86400s 以控 Redis 内存) - metrics: 加 userPlatformQuotaSentinelSetCacheErrorTotal,并入 GatewayUserPlatformQuotaIncrStats 暴露 - billing_cache: checkUserPlatformQuotaEligibility 在 cache MISS + DB 无行且 cacheErr==nil 时回填 sentinel(三 limit nil、三 window_start non-nil、SchemaV1); TTL<=0 fallback 1h 防 EXPIRE 立即删 key 击穿;SET 失败 fail-open + 计 metric - billing_cache: HIT 路径对 sentinel(三 limit nil)跳过 windowExpired refresh, 避免短 sentinel TTL 被误升级为 86400s 有配额 limit 的用户 enforcement 行为不变(rec!=nil 不回填、isSentinel=false 不跳过 refresh)。 测试:扩展 fakeFullCache 夹具(setCalls/lastSetTTL/getErr/setErr);新增回填正确性 / Redis-GET-故障不回填 / SET-失败 fail-open / sentinel 跨窗口不 refresh 四个单测。 go build、quota+billing unit、三态 go vet 全绿。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
DaydreamCoding
parent
7321e4dea8
commit
06fca66273
@ -649,6 +649,9 @@ type BillingConfig struct {
|
||||
// - billing_cache_service.checkUserPlatformQuotaEligibility 首次缓存装载
|
||||
// 读写两端必须共用同一 TTL，避免缓存生命周期不一致导致 quota 计数漂移。
|
||||
UserPlatformQuotaCacheTTLSeconds int `mapstructure:"user_platform_quota_cache_ttl_seconds"`
|
||||
// UserPlatformQuotaSentinelTTLSeconds sentinel(无 limit 占位)entry 的 TTL,
|
||||
// 显著短于 quota cache 默认 86400s 以控 Redis 内存;默认 3600=1h。
|
||||
UserPlatformQuotaSentinelTTLSeconds int `mapstructure:"user_platform_quota_sentinel_ttl_seconds"`
|
||||
}
|
||||
|
||||
type CircuitBreakerConfig struct {
|
||||
@ -1571,6 +1574,7 @@ func setDefaults() {
|
||||
viper.SetDefault("billing.circuit_breaker.reset_timeout_seconds", 30)
|
||||
viper.SetDefault("billing.circuit_breaker.half_open_requests", 3)
|
||||
viper.SetDefault("billing.user_platform_quota_cache_ttl_seconds", 86400)
|
||||
viper.SetDefault("billing.user_platform_quota_sentinel_ttl_seconds", 3600)
|
||||
|
||||
// Turnstile
|
||||
viper.SetDefault("turnstile.required", false)
|
||||
|
||||
@ -1096,7 +1096,12 @@ func (s *BillingCacheService) checkUserPlatformQuotaEligibility(
|
||||
// 超时 50ms:覆盖正常路径与可接受抖动;Redis 异常时 hot path 不阻塞超过此值。
|
||||
// 用 context.Background()+短超时,避免请求 ctx 取消导致刷新丢失。
|
||||
// 显式 setCancel()(而非 defer):缩短 context 生命周期,避免 defer 延迟到函数返回。
|
||||
if windowExpired && s.cache != nil {
|
||||
// isSentinel 判定「该 entry 无任何 limit」,涵盖两类,跨窗口命中时都跳过 refresh:
|
||||
// 1) A3 回填的 sentinel(DB 无行,短 TTL):refresh 会把短 TTL 误升级为 86400s,有害;
|
||||
// 2) DB 有行但三 limit 全未配置的用户(TTL 86400s):refresh 纯属无意义(TTL 升级本身无害)。
|
||||
// 两类的 enforcement(下方 limit!=nil 比较)都因 limit 全 nil 永远放行,跳过 refresh 均正确。
|
||||
isSentinel := entry.DailyLimitUSD == nil && entry.WeeklyLimitUSD == nil && entry.MonthlyLimitUSD == nil
|
||||
if windowExpired && s.cache != nil && !isSentinel {
|
||||
refreshed := &UserPlatformQuotaCacheEntry{
|
||||
DailyUsageUSD: dailyUsage,
|
||||
WeeklyUsageUSD: weeklyUsage,
|
||||
@ -1159,6 +1164,33 @@ func (s *BillingCacheService) checkUserPlatformQuotaEligibility(
|
||||
}
|
||||
rec, _ := v.(*UserPlatformQuotaRecord)
|
||||
if rec == nil {
|
||||
// 仅在 cache 可用且本次 GET 未出错时回填 sentinel:Redis GET 故障(cacheErr!=nil)
|
||||
// 时不回填,与下方 line ~1201 "Redis 故障时 fail-open:不回填" 保持一致,
|
||||
// 避免在 Redis 异常期做一次注定失败的 SET。
|
||||
if s.cache != nil && cacheErr == nil {
|
||||
now := time.Now()
|
||||
startOfDay := timezone.StartOfDay(now)
|
||||
startOfWeek := timezone.StartOfWeek(now)
|
||||
sentinel := &UserPlatformQuotaCacheEntry{
|
||||
SchemaVersion: UserPlatformQuotaCacheSchemaV1,
|
||||
DailyWindowStart: &startOfDay,
|
||||
WeeklyWindowStart: &startOfWeek,
|
||||
MonthlyWindowStart: &now,
|
||||
// limits 全 nil, usage 全 0(零值)
|
||||
}
|
||||
sentinelTTL := time.Duration(s.cfg.Billing.UserPlatformQuotaSentinelTTLSeconds) * time.Second
|
||||
if sentinelTTL <= 0 {
|
||||
// 防御:TTL<=0 时 Redis EXPIRE 会立即删除整个 key(见 billing_cache.go 的 pipe.Expire),
|
||||
// sentinel 不持久化 → 每请求击穿 DB。配置缺失/误配为 0 时 fallback 到 1h。
|
||||
sentinelTTL = time.Hour
|
||||
}
|
||||
setCtx, setCancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
||||
if setErr := s.cache.SetUserPlatformQuotaCache(setCtx, userID, platform, sentinel, sentinelTTL); setErr != nil {
|
||||
userPlatformQuotaSentinelSetCacheErrorTotal.Add(1)
|
||||
logger.LegacyPrintf("service.billing_cache", "Warning: set sentinel quota cache failed user=%d platform=%s: %v", userID, platform, setErr)
|
||||
}
|
||||
setCancel()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@ -95,6 +95,10 @@ type fakeFullCache struct {
|
||||
mu sync.Mutex
|
||||
entry *UserPlatformQuotaCacheEntry
|
||||
deleteCalls int
|
||||
setCalls int // SetUserPlatformQuotaCache 调用次数
|
||||
lastSetTTL time.Duration // 最近一次 Set 的 ttl
|
||||
getErr error // 非 nil 时 Get 先返回 (nil,false,getErr)
|
||||
setErr error // 非 nil 时 Set 返回该 err(setCalls 仍+1)
|
||||
}
|
||||
|
||||
// getDeleteCalls 线程安全地读取 deleteCalls。
|
||||
@ -111,19 +115,41 @@ func (f *fakeFullCache) getEntry() *UserPlatformQuotaCacheEntry {
|
||||
return f.entry
|
||||
}
|
||||
|
||||
// getSetCalls 线程安全地读取 setCalls。
|
||||
func (f *fakeFullCache) getSetCalls() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.setCalls
|
||||
}
|
||||
|
||||
// getLastSetTTL 线程安全地读取 lastSetTTL。
|
||||
func (f *fakeFullCache) getLastSetTTL() time.Duration {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.lastSetTTL
|
||||
}
|
||||
|
||||
func (f *fakeFullCache) GetUserPlatformQuotaCache(_ context.Context, _ int64, _ string) (*UserPlatformQuotaCacheEntry, bool, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
if f.getErr != nil {
|
||||
return nil, false, f.getErr
|
||||
}
|
||||
if f.entry == nil {
|
||||
return nil, false, nil
|
||||
}
|
||||
return f.entry, true, nil
|
||||
}
|
||||
|
||||
func (f *fakeFullCache) SetUserPlatformQuotaCache(_ context.Context, _ int64, _ string, e *UserPlatformQuotaCacheEntry, _ time.Duration) error {
|
||||
func (f *fakeFullCache) SetUserPlatformQuotaCache(_ context.Context, _ int64, _ string, e *UserPlatformQuotaCacheEntry, ttl time.Duration) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.setCalls++
|
||||
if f.setErr != nil {
|
||||
return f.setErr
|
||||
}
|
||||
f.entry = e
|
||||
f.lastSetTTL = ttl
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -593,3 +619,97 @@ func TestMonthlyQuotaWindowExpired_BoundaryTable(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckUserPlatformQuotaEligibility_NoRow_WritesSentinel 验证:
|
||||
// cache MISS + DB 无行时,回填 sentinel entry(三 limit 全 nil,三 window_start 全 non-nil),
|
||||
// TTL = UserPlatformQuotaSentinelTTLSeconds,函数返回 nil(fail-open)。
|
||||
func TestCheckUserPlatformQuotaEligibility_NoRow_WritesSentinel(t *testing.T) {
|
||||
repo := &fakeQuotaRepo{rec: nil} // DB 无行
|
||||
cache := &fakeFullCache{} // entry=nil → Get 返回 MISS
|
||||
svc := newServiceForPreflight(t, repo, cache)
|
||||
svc.cfg.Billing.UserPlatformQuotaSentinelTTLSeconds = 3600
|
||||
|
||||
if err := svc.checkUserPlatformQuotaEligibility(context.Background(), 1, "anthropic"); err != nil {
|
||||
t.Fatalf("expected nil (fail-open), got %v", err)
|
||||
}
|
||||
if cache.getSetCalls() != 1 {
|
||||
t.Fatalf("expected 1 SetUserPlatformQuotaCache call for sentinel, got %d", cache.getSetCalls())
|
||||
}
|
||||
sentinel := cache.getEntry()
|
||||
if sentinel == nil {
|
||||
t.Fatal("expected sentinel entry backfilled")
|
||||
}
|
||||
if sentinel.DailyLimitUSD != nil || sentinel.WeeklyLimitUSD != nil || sentinel.MonthlyLimitUSD != nil {
|
||||
t.Errorf("sentinel must have all-nil limits")
|
||||
}
|
||||
if sentinel.DailyWindowStart == nil || sentinel.WeeklyWindowStart == nil || sentinel.MonthlyWindowStart == nil {
|
||||
t.Errorf("sentinel must have non-nil window_start to avoid refresh churn")
|
||||
}
|
||||
if sentinel.SchemaVersion != UserPlatformQuotaCacheSchemaV1 {
|
||||
t.Errorf("sentinel schema = %d, want V1", sentinel.SchemaVersion)
|
||||
}
|
||||
if cache.getLastSetTTL() != 3600*time.Second {
|
||||
t.Errorf("sentinel ttl = %v, want 3600s", cache.getLastSetTTL())
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckUserPlatformQuotaEligibility_RedisGetError_NoSentinelBackfill 验证:
|
||||
// Redis GET 故障(cacheErr!=nil)+ DB 无行时,不应回填 sentinel(与 "Redis 故障时不回填" 一致),且 fail-open。
|
||||
func TestCheckUserPlatformQuotaEligibility_RedisGetError_NoSentinelBackfill(t *testing.T) {
|
||||
repo := &fakeQuotaRepo{rec: nil}
|
||||
cache := &fakeFullCache{getErr: errors.New("redis get down")}
|
||||
svc := newServiceForPreflight(t, repo, cache)
|
||||
svc.cfg.Billing.UserPlatformQuotaSentinelTTLSeconds = 3600
|
||||
|
||||
if err := svc.checkUserPlatformQuotaEligibility(context.Background(), 1, "anthropic"); err != nil {
|
||||
t.Fatalf("redis 故障应 fail-open, got %v", err)
|
||||
}
|
||||
if cache.getSetCalls() != 0 {
|
||||
t.Errorf("redis-get-error 时不应回填 sentinel, got %d set calls", cache.getSetCalls())
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckUserPlatformQuotaEligibility_NoRow_SentinelSetFailsFailOpen 验证:
|
||||
// sentinel SET 失败时 fail-open(返回 nil)且计 metric。
|
||||
func TestCheckUserPlatformQuotaEligibility_NoRow_SentinelSetFailsFailOpen(t *testing.T) {
|
||||
before := userPlatformQuotaSentinelSetCacheErrorTotal.Load()
|
||||
repo := &fakeQuotaRepo{rec: nil}
|
||||
cache := &fakeFullCache{setErr: errors.New("redis set timeout")}
|
||||
svc := newServiceForPreflight(t, repo, cache)
|
||||
svc.cfg.Billing.UserPlatformQuotaSentinelTTLSeconds = 3600
|
||||
|
||||
if err := svc.checkUserPlatformQuotaEligibility(context.Background(), 1, "anthropic"); err != nil {
|
||||
t.Fatalf("sentinel set 失败应 fail-open, got %v", err)
|
||||
}
|
||||
if cache.getSetCalls() != 1 {
|
||||
t.Errorf("应尝试 set sentinel 恰好一次, got %d", cache.getSetCalls())
|
||||
}
|
||||
if got := userPlatformQuotaSentinelSetCacheErrorTotal.Load() - before; got != 1 {
|
||||
t.Errorf("set 失败应使 metric +1, got delta %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckUserPlatformQuotaEligibility_SentinelCrossDay_NoRefresh 验证:
|
||||
// 命中 sentinel(三 limit 全 nil)且跨窗口(daily/weekly 过期)时,不应触发 refresh SetCache
|
||||
// (否则会把短 sentinel TTL 误升级为 quota cache 默认 86400s)。
|
||||
func TestCheckUserPlatformQuotaEligibility_SentinelCrossDay_NoRefresh(t *testing.T) {
|
||||
yesterday := timezone.StartOfDay(time.Now().AddDate(0, 0, -1))
|
||||
lastWeek := timezone.StartOfWeek(time.Now().AddDate(0, 0, -7))
|
||||
monthAgoOK := time.Now().AddDate(0, 0, -5) // <30d, monthly 不过期
|
||||
sentinel := &UserPlatformQuotaCacheEntry{
|
||||
SchemaVersion: UserPlatformQuotaCacheSchemaV1,
|
||||
DailyWindowStart: &yesterday, // 跨日 → daily windowExpired = true
|
||||
WeeklyWindowStart: &lastWeek, // 跨周 → weekly windowExpired = true
|
||||
MonthlyWindowStart: &monthAgoOK,
|
||||
// limits 全 nil → sentinel
|
||||
}
|
||||
cache := &fakeFullCache{entry: sentinel} // entry 非 nil → Get HIT
|
||||
svc := newServiceForPreflight(t, &fakeQuotaRepo{}, cache)
|
||||
|
||||
if err := svc.checkUserPlatformQuotaEligibility(context.Background(), 1, "anthropic"); err != nil {
|
||||
t.Fatalf("sentinel = no limit, expected nil, got %v", err)
|
||||
}
|
||||
if cache.getSetCalls() != 0 {
|
||||
t.Errorf("sentinel cross-window must NOT trigger refresh SetCache, got %d calls", cache.getSetCalls())
|
||||
}
|
||||
}
|
||||
|
||||
@ -105,6 +105,9 @@ var (
|
||||
// （applyUsageBilling 在 repo==nil 时 fallback）路径下的失败次数；
|
||||
// 与 DB Incr 失败分开计数，便于区分"主路径暂时故障"vs"基础设施长期未配齐"。
|
||||
userPlatformQuotaDBIncrLegacyErrorTotal atomic.Int64
|
||||
// userPlatformQuotaSentinelSetCacheErrorTotal 统计 checkUserPlatformQuotaEligibility
|
||||
// 在 DB 无行时回填 sentinel cache entry 写 Redis 失败的次数（phase A）。
|
||||
userPlatformQuotaSentinelSetCacheErrorTotal atomic.Int64
|
||||
)
|
||||
|
||||
func GatewayWindowCostPrefetchStats() (cacheHit, cacheMiss, batchSQL, fallback, errCount int64) {
|
||||
@ -127,13 +130,15 @@ func GatewayModelsListCacheStats() (cacheHit, cacheMiss, store int64) {
|
||||
return modelsListCacheHitTotal.Load(), modelsListCacheMissTotal.Load(), modelsListCacheStoreTotal.Load()
|
||||
}
|
||||
|
||||
// GatewayUserPlatformQuotaIncrStats 返回 (mainPathErr, legacyPathErr)。
|
||||
// GatewayUserPlatformQuotaIncrStats 返回 (mainPathErr, legacyPathErr, sentinelSetErr)。
|
||||
// mainPathErr：finalizePostUsageBilling 异步 goroutine 写 DB 失败累计次数；
|
||||
// legacyPathErr：postUsageBilling fallback 路径写 DB 失败累计次数。
|
||||
// legacyPathErr：postUsageBilling fallback 路径写 DB 失败累计次数；
|
||||
// sentinelSetErr：DB 无行时回填 sentinel cache entry 写 Redis 失败累计次数。
|
||||
// ops 监控面板可以按"持续上升斜率"做告警阈值。
|
||||
func GatewayUserPlatformQuotaIncrStats() (mainPathErr, legacyPathErr int64) {
|
||||
func GatewayUserPlatformQuotaIncrStats() (mainPathErr, legacyPathErr, sentinelSetErr int64) {
|
||||
return userPlatformQuotaDBIncrErrorTotal.Load(),
|
||||
userPlatformQuotaDBIncrLegacyErrorTotal.Load()
|
||||
userPlatformQuotaDBIncrLegacyErrorTotal.Load(),
|
||||
userPlatformQuotaSentinelSetCacheErrorTotal.Load()
|
||||
}
|
||||
|
||||
func openAIStreamEventIsTerminal(data string) bool {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user