seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-14 03:52:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(gemini): 显式缓存恢复流程区分账号已删除与暂不支持缓存，上游必已过期后作废意图

Browse Source 
后台回滚在持有账号返回 nil 时把修改意图当作已回滚清除：账号临时关闭显式缓存能力后
意图丢失，重新开启也无法补偿。

- boundAccount 区分两种情形：账号已删除返回 nil，账号存在但不支持显式缓存返回
  errGeminiCachedContentAccountUnsupported；BoundAccount 对外行为不变
- 账号暂不支持缓存时不改动上游、保留意图，租约过期后重试（期间同一缓存的修改返回 409）；
  账号已删除时没有凭据可回滚上游，软删本地记录
- 修改前到期时间加上单次最长有效期与 1 小时容忍都已过去时上游缓存必已自行过期：
  账号不支持缓存或上游回滚持续失败的意图不再重试，直接作废本地记录
- 测试夹具 GetForOwner 过滤已软删记录
This commit is contained in:
feeeei
2026-10-13 00:54:27 +08:00
parent 26b31a83e4
commit 43a3c197cb
2 changed files with 217 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/handler/gemini_cached_contents_billing_test.go
+164 -1
View File
@@ -62,7 +62,7 @@ func (r *cachedBillingRepo) GetForOwner(_ context.Context, apiKeyID int64, publi
r.mu.Lock()
defer r.mu.Unlock()
record, ok := r.records[publicID]
if !ok || record.APIKeyID != apiKeyID {
if !ok || record.APIKeyID != apiKeyID || r.liveLocked(record.ID) == nil {
return nil, service.ErrGeminiCachedContentNotFound
}
cp := *record
@@ -790,3 +790,166 @@ func TestGeminiCachedContentsDelete_UpstreamNotFoundStillForgets(t *testing.T) {
})
}
}
// 持有账号暂时不支持显式缓存（能力被关闭）时，恢复流程不能把修改意图当作已回滚而清除；
// 能力恢复后再回滚上游并清除意图。
func TestGeminiCachedContentMutationRecovery_KeepsIntentWhileAccountCannotCache(t *testing.T) {
env := newCachedBillingEnv(t, 1000, 900)
env.accounts[900].Credentials["base_url"] = "https://gw.example.com"
env.accounts[900].Credentials[service.GeminiExplicitCacheUpstreamCredentialKey] = true
baseline := time.Now().Add(time.Hour)
env.seed(baseline)
env.repo.failCommit = true
require.Equal(t, http.StatusInternalServerError, env.patch().Code)
require.True(t, env.repo.pendingMutation(100))
require.Equal(t, 1, env.upstream.methodCalls(http.MethodPatch))
env.repo.failCommit = false
env.repo.expireLease(100)
env.accounts[900].Credentials[service.GeminiExplicitCacheUpstreamCredentialKey] = false
recovered, err := env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.Error(t, err, "账号暂不支持缓存时恢复失败")
require.Zero(t, recovered)
require.True(t, env.repo.pendingMutation(100), "未回滚上游前保留修改意图")
require.Equal(t, 1, env.upstream.methodCalls(http.MethodPatch), "不改动上游")
require.Empty(t, env.repo.deleted)
env.accounts[900].Credentials[service.GeminiExplicitCacheUpstreamCredentialKey] = true
env.repo.expireLease(100)
recovered, err = env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.NoError(t, err)
require.Equal(t, 1, recovered)
require.Equal(t, 2, env.upstream.methodCalls(http.MethodPatch))
requireTTLNear(t, env.upstream.patchTTLs[1], time.Until(baseline))
require.False(t, env.repo.pendingMutation(100))
require.True(t, env.repo.records["abc"].ExpireTime.Equal(baseline))
require.Empty(t, env.usage.totals())
}
// 持有账号长期不支持显式缓存，且修改前到期时间加上单次最长有效期都已过去：上游缓存必已自行过期，
// 恢复流程不再保留意图，直接移除本地记录。
func TestGeminiCachedContentMutationRecovery_AbandonsIntentOnceUpstreamMustHaveExpired(t *testing.T) {
env := newCachedBillingEnv(t, 1000, 900)
env.accounts[900].Credentials["base_url"] = "https://gw.example.com"
env.accounts[900].Credentials[service.GeminiExplicitCacheUpstreamCredentialKey] = true
env.seed(time.Now().Add(time.Hour))
env.repo.failCommit = true
require.Equal(t, http.StatusInternalServerError, env.patch().Code)
env.repo.failCommit = false
env.repo.expireLease(100)
env.accounts[900].Credentials[service.GeminiExplicitCacheUpstreamCredentialKey] = false
// 修改前到期时间 + 24h 上限 + 1h 容忍仍未过去：保留意图。
env.repo.records["abc"].ExpireTime = time.Now().Add(-20 * time.Hour)
recovered, err := env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.Error(t, err)
require.Zero(t, recovered)
require.True(t, env.repo.pendingMutation(100))
require.Empty(t, env.repo.deleted)
env.repo.expireLease(100)
env.repo.records["abc"].ExpireTime = time.Now().Add(-26 * time.Hour)
recovered, err = env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.NoError(t, err)
require.Equal(t, 1, recovered)
require.Equal(t, 1, env.upstream.methodCalls(http.MethodPatch), "不改动上游")
require.Zero(t, env.upstream.methodCalls(http.MethodDelete))
require.Equal(t, []int64{100}, env.repo.deleted)
require.Empty(t, env.usage.totals())
recovered, err = env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.NoError(t, err)
require.Zero(t, recovered, "作废的记录不再被选中")
}
// 作废判定的边界：默认上限 24h 加 1h 容忍；配置上限低于 24h 时仍按 24h 兜底。
func TestGeminiCachedContentMutationRecovery_AbandonBoundary(t *testing.T) {
disableCache := func(env *cachedBillingEnv) {
env.accounts[900].Credentials["base_url"] = "https://gw.example.com"
env.accounts[900].Credentials[service.GeminiExplicitCacheUpstreamCredentialKey] = false
}
t.Run("within grace keeps intent", func(t *testing.T) {
env := newCachedBillingEnv(t, 1000, 900)
env.seed(time.Now().Add(-25*time.Hour + time.Minute))
env.repo.mutations[100] = cachedBillingMutation{token: "stale", leaseUntil: time.Now().Add(-time.Second)}
disableCache(env)
recovered, err := env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.Error(t, err)
require.Zero(t, recovered)
require.True(t, env.repo.pendingMutation(100))
require.Empty(t, env.repo.deleted)
})
t.Run("configured max ttl below default still waits a day", func(t *testing.T) {
env := newCachedBillingEnv(t, 1000, 900)
env.seed(time.Now().Add(-12 * time.Hour))
env.repo.mutations[100] = cachedBillingMutation{token: "stale", leaseUntil: time.Now().Add(-time.Second)}
disableCache(env)
cfg := &config.Config{}
cfg.Gateway.GeminiCachedContentMaxTTLSeconds = 3600
compat := service.NewGeminiMessagesCompatService(&cachedBillingAccountRepo{accounts: env.accounts}, nil, nil, nil, nil, nil, env.upstream, nil, cfg)
svc := service.NewGeminiCachedContentService(env.repo, compat, cfg)
recovered, err := svc.RecoverStaleMutations(context.Background())
require.Error(t, err)
require.Zero(t, recovered)
require.True(t, env.repo.pendingMutation(100))
env.repo.expireLease(100)
env.repo.records["abc"].ExpireTime = time.Now().Add(-26 * time.Hour)
recovered, err = svc.RecoverStaleMutations(context.Background())
require.NoError(t, err)
require.Equal(t, 1, recovered)
require.Equal(t, []int64{100}, env.repo.deleted)
})
}
// 持有账号可用但上游回滚持续失败（如 Key 已失效）：上游缓存必已自行过期后不再重试，移除本地记录。
func TestGeminiCachedContentMutationRecovery_AbandonsIntentWhenRollbackKeepsFailing(t *testing.T) {
env := newCachedBillingEnv(t, 1000, 900)
env.seed(time.Now().Add(-time.Minute))
env.repo.mutations[100] = cachedBillingMutation{token: "stale", leaseUntil: time.Now().Add(-time.Second)}
env.upstream.failures = map[string]cachedBillingUpstreamFailure{http.MethodDelete: {status: http.StatusUnauthorized,
body: `{"error":{"code":401,"message":"API key not valid","status":"UNAUTHENTICATED"}}`}}
recovered, err := env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.Error(t, err, "上游缓存可能仍存活：保留意图等待重试")
require.Zero(t, recovered)
require.Equal(t, 1, env.upstream.methodCalls(http.MethodDelete))
require.True(t, env.repo.pendingMutation(100))
require.Empty(t, env.repo.deleted)
env.repo.expireLease(100)
env.repo.records["abc"].ExpireTime = time.Now().Add(-26 * time.Hour)
recovered, err = env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.NoError(t, err)
require.Equal(t, 1, recovered)
require.Equal(t, 2, env.upstream.methodCalls(http.MethodDelete), "账号可用时仍先尝试回滚")
require.Equal(t, []int64{100}, env.repo.deleted)
require.Empty(t, env.usage.totals())
}
// 持有账号已删除时没有凭据可回滚上游：恢复流程移除本地记录并清除意图，不再重试。
func TestGeminiCachedContentMutationRecovery_ForgetsRecordWhenAccountDeleted(t *testing.T) {
env := newCachedBillingEnv(t, 1000, 900)
env.seed(time.Now().Add(time.Hour))
env.repo.failCommit = true
require.Equal(t, http.StatusInternalServerError, env.patch().Code)
require.True(t, env.repo.pendingMutation(100))
env.repo.failCommit = false
env.repo.expireLease(100)
delete(env.accounts, 900)
recovered, err := env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.NoError(t, err)
require.Equal(t, 1, recovered)
require.Equal(t, 1, env.upstream.methodCalls(http.MethodPatch), "无凭据，不改动上游")
require.Zero(t, env.upstream.methodCalls(http.MethodDelete))
require.Equal(t, []int64{100}, env.repo.deleted, "本地记录作废")
require.Empty(t, env.usage.totals())
require.Equal(t, http.StatusForbidden, env.serve(http.MethodGet, "/v1beta/cachedContents/abc", "", gin.Params{{Key: "cacheID", Value: "abc"}}, env.h.GeminiCachedContentsGet).Code, "作废的记录对客户端不可见")
recovered, err = env.h.geminiCachedContentService.RecoverStaleMutations(context.Background())
require.NoError(t, err)
require.Zero(t, recovered, "不再重复处理")
}
backend/internal/service/gemini_cached_content.go
+53 -5
View File
@@ -54,6 +54,8 @@ const (
geminiCachedContentMutationRecoveryInterval = time.Minute
geminiCachedContentMutationRecoveryBatch = 50
geminiCachedContentMutationRecoveryTimeout = time.Minute
// geminiCachedContentMutationAbandonGrace 是判定"上游缓存必已自行过期"时额外容忍的网关与上游时钟偏差。
geminiCachedContentMutationAbandonGrace = time.Hour
// GeminiCachedContentListDefaultPageSize / GeminiCachedContentListMaxPageSize 约束 list 分页。
GeminiCachedContentListDefaultPageSize = 50
@@ -67,6 +69,8 @@ var (
ErrGeminiCachedContentMutationConflict = infraerrors.New(http.StatusConflict, "GEMINI_CACHED_CONTENT_MUTATION_CONFLICT", "another update to this cached content is in progress")
// ErrGeminiCachedContentRequestNotSent 表示上游缓存请求在发出前失败（地址、凭据等），上游状态未被改变。
ErrGeminiCachedContentRequestNotSent = errors.New("gemini cached content request was not sent")
// errGeminiCachedContentAccountUnsupported 表示持有缓存的账号存在，但当前不支持显式缓存（能力被关闭或认证方式不支持）。
errGeminiCachedContentAccountUnsupported = errors.New("gemini cached content account does not support explicit caching")
geminiCachedContentPublicIDPattern = regexp.MustCompile(`^[a-z0-9]{1,64}$`)
geminiCachedContentAIStudioName = regexp.MustCompile(`^cachedContents/[A-Za-z0-9_-]+$`)
@@ -815,7 +819,8 @@ func (s *GeminiCachedContentService) RecoverStaleMutations(ctx context.Context)
}
// recoverStaleMutation 接管 id 上租约已过期的修改意图，把上游有效期回滚到库中的到期时间后清除意图；
// 库中到期时间已到时删除上游缓存与本地记录。没有可接管的意图时返回 false；回滚失败时意图保留，租约过期后重试。
// 库中到期时间已到时删除上游缓存与本地记录。没有可接管的意图时返回 false；回滚失败时意图保留，
// 本次接管取得的新租约（geminiCachedContentMutationLease）过期后重试，期间同一缓存的修改返回 409。
func (s *GeminiCachedContentService) recoverStaleMutation(ctx context.Context, id int64) (bool, error) {
token, err := NewGeminiCachedContentPublicID()
if err != nil {
@@ -836,12 +841,25 @@ func (s *GeminiCachedContentService) recoverStaleMutation(ctx context.Context, i
}
// rollbackUpstream 把上游有效期改回 record.ExpireTime，剩余不足 1 秒时删除上游缓存；
// 返回 true 表示上游缓存已不存在，本地记录应移除。持有账号已不可用时不改动上游。
// 返回 true 表示上游缓存已不存在、或已没有凭据可触达（持有账号已删除），本地记录应移除。
// 持有账号存在但当前不支持显式缓存时不改动上游，修改意图保留到账号恢复后重试；账号可用时总是先向上游回滚。
// 两种情况下只要 mutationUpstreamMustHaveExpired 成立（上游缓存必已自行过期），本地记录直接作废，不再重试。
func (s *GeminiCachedContentService) rollbackUpstream(ctx context.Context, record *GeminiCachedContent) (bool, error) {
account, err := s.BoundAccount(ctx, record)
if err != nil || account == nil {
account, err := s.boundAccount(ctx, record)
if errors.Is(err, errGeminiCachedContentAccountUnsupported) {
if s.mutationUpstreamMustHaveExpired(record) {
logger.LegacyPrintf("service.gemini_cached_content", "Warning: gemini cached content %d (upstream %s) has a pending mutation but account %d stopped supporting explicit caching and the upstream cache has expired since; dropping the local record", record.ID, record.UpstreamName, record.AccountID)
return true, nil
}
return false, fmt.Errorf("%w (account %d)", err, record.AccountID)
}
if err != nil {
return false, err
}
if account == nil {
logger.LegacyPrintf("service.gemini_cached_content", "Warning: gemini cached content %d (upstream %s) has a pending mutation but account %d no longer exists; dropping the local record without touching upstream", record.ID, record.UpstreamName, record.AccountID)
return true, nil
}
remaining := record.ExpireTime.Sub(s.now())
if remaining < time.Second {
err = s.DeleteUpstream(ctx, account, record)
@@ -853,17 +871,44 @@ func (s *GeminiCachedContentService) rollbackUpstream(ctx context.Context, recor
return true, nil
}
if err != nil {
if s.mutationUpstreamMustHaveExpired(record) {
logger.LegacyPrintf("service.gemini_cached_content", "Warning: gemini cached content %d (upstream %s) rollback keeps failing on account %d but the upstream cache has expired since; dropping the local record: %v", record.ID, record.UpstreamName, record.AccountID, err)
return true, nil
}
return false, err
}
return remaining < time.Second, nil
}
// mutationUpstreamMustHaveExpired 判断 record 上未完成修改所涉及的上游缓存是否必已自行过期：
// 修改发生时记录尚未到期（Resolve 拒绝已过期记录），且单次延长不超过 MaxTTL，
// 所以上游有效期最晚为 record.ExpireTime + MaxTTL；再过 geminiCachedContentMutationAbandonGrace 后判定为已过期。
// MaxTTL 取判定时刻的配置并以默认上限兜底：配置被调低后判定可能早于上游真实过期，此时只影响运营方的上游存储费，
// 不涉及用户计费。
func (s *GeminiCachedContentService) mutationUpstreamMustHaveExpired(record *GeminiCachedContent) bool {
maxTTL := s.MaxTTL()
if maxTTL < geminiCachedContentDefaultMaxTTL {
maxTTL = geminiCachedContentDefaultMaxTTL
}
return s.now().After(record.ExpireTime.Add(maxTTL + geminiCachedContentMutationAbandonGrace))
}
func (s *GeminiCachedContentService) Forget(ctx context.Context, record *GeminiCachedContent) error {
return s.repo.SoftDelete(ctx, record.ID)
}
// BoundAccount 返回持有缓存的账号；账号已删除或不再支持显式缓存时返回 nil。
func (s *GeminiCachedContentService) BoundAccount(ctx context.Context, record *GeminiCachedContent) (*Account, error) {
account, err := s.boundAccount(ctx, record)
if errors.Is(err, errGeminiCachedContentAccountUnsupported) {
return nil, nil
}
return account, err
}
// boundAccount 返回持有缓存的账号：账号已删除返回 nil；账号存在但当前不支持显式缓存返回
// errGeminiCachedContentAccountUnsupported，调用方据此区分"永久不可用"与"暂时不可用"。
func (s *GeminiCachedContentService) boundAccount(ctx context.Context, record *GeminiCachedContent) (*Account, error) {
if s.compat == nil || s.compat.accountRepo == nil {
return nil, errors.New("account repository not configured")
}
@@ -874,9 +919,12 @@ func (s *GeminiCachedContentService) BoundAccount(ctx context.Context, record *G
}
return nil, err
}
if account == nil || !IsGeminiExplicitCacheAccount(account) {
if account == nil {
return nil, nil
}
if !IsGeminiExplicitCacheAccount(account) {
return nil, errGeminiCachedContentAccountUnsupported
}
return account, nil
}