fix(batch-image): 修复审计发现的计费死锁、状态机与队列原子性缺陷

修复 PR #3768 批量图像 MVP 合并后审计报告中的全部问题:

结算与计费(高危):
- 所有 SETTLEMENT_* 失败(超冻结/计数非法/manifest 冲突/定价缺失/扣费失败)
  统一计入 retry_count 并在耗尽时释放冻结转 failed,消灭 settling 无限
  requeue 导致的冻结余额永久锁死
- 耗尽出口的释放指纹统一为 RequestHash,与 processor/Cancel/recovery 一致;
  release 遇同 request id 指纹冲突视为幂等成功,治愈历史毒消息
- 管理端校验 hold_multiplier >= discount_multiplier,定价快照对存量脏数据钳制
- 释放前校验 per-job hold claim(dedup+归档表),杜绝幻影释放

索引对账(高危):
- provider 输出与提交 custom_id 集对账:未知条目丢弃并记事件,
  漏项补 PROVIDER_RESULT_MISSING 失败行,保证 success+fail == item_count

提交与恢复(高危):
- 提交前转 uploading 并在 provider.Submit 期间心跳刷新 updated_at;
  恢复扫描改为原子复核(FailStaleUnsubmittedBatchImageJob),
  消灭慢提交被误杀退款而上游任务照常计费的孤儿场景
- 上游任务创建成功但本地状态推进失败时,尽力取消上游并清理输入
- recovery 释放失败时入队交由 worker releaseTerminalHold 兜底重试

队列与并发(中危):
- Enqueue(SetNX+LPush)与 Reserve(BRPop+ZAdd)均改为 Lua 原子脚本,
  消灭崩溃窗口导致 job 脱离队列、被 7 天 inflight 键锁死
- 锁冲突按 LockConflictDelay 重新入队(原直接丢弃需等 10 分钟 stale 恢复)
- 处理期间心跳:active zset 续期(ZAddXX 防幽灵成员)+ 锁 TTL 续期
- ReplaceBatchImageItemsForJob 增加 indexing 状态守卫,防掉队 worker 重写账目

存量回归(中危):
- image-only 定价条目(仅图片价无 token 价)恢复 token 计费 fail-closed,
  不再按 $0 计费;图片计费路径不受影响
- 鉴权余额门槛恢复 balance <= 0 语义,MinimumBalanceReserve 不再作硬 403

加固:
- ZIP max_items 钳制到管理员上限;Submit 补齐 Platform==Gemini 校验;
  gemini downloadUri 跟随前做 host 白名单校验
- 批量客户端改用共享 httpclient(拨号/TLS/响应头超时有界)
- 审计点名的忽略错误(MarkDownloaded/SettlementFailed/AppendEvent 等)改为记日志
This commit is contained in:
shaw
2026-07-07 18:53:56 +08:00
parent 44ab690a01
commit 80a229bce5
31 changed files with 1180 additions and 90 deletions
@@ -289,14 +289,11 @@ func setGroupContext(c *gin.Context, group *service.Group) {
c.Request = c.Request.WithContext(ctx)
}
func apiKeyBalanceBelowAuthThreshold(balance float64, cfg *config.Config) bool {
if balance <= 0 {
return true
}
if cfg == nil || cfg.Billing.MinimumBalanceReserve <= 0 {
return false
}
return balance < cfg.Billing.MinimumBalanceReserve
// apiKeyBalanceBelowAuthThreshold 保持鉴权层的历史语义:仅在余额耗尽(<=0)时拒绝。
// MinimumBalanceReserve 只作为 billing-cache 预检的保守下限,不得复用为鉴权硬门槛,
// 否则已配置该值的存量部署升级后,0 < balance < reserve 的用户会在所有端点被静默 403。
func apiKeyBalanceBelowAuthThreshold(balance float64, _ *config.Config) bool {
return balance <= 0
}
func abortIfAPIKeyGroupUnavailable(c *gin.Context, apiKey *service.APIKey) bool {
@@ -542,6 +542,8 @@ func TestApiKeyAuthWithSubscriptionGoogle_InsufficientBalance(t *testing.T) {
func TestApiKeyAuthWithSubscriptionGoogle_BalanceBelowMinimumReserve(t *testing.T) {
gin.SetMode(gin.TestMode)
// 鉴权层保持历史语义:MinimumBalanceReserve 只用于 billing-cache 预检,
// 0 < balance < reserve 的用户不得在鉴权中间件被硬 403。
r := gin.New()
apiKeyService := newTestAPIKeyService(fakeAPIKeyRepo{
getByKey: func(ctx context.Context, key string) (*service.APIKey, error) {
@@ -567,6 +569,36 @@ func TestApiKeyAuthWithSubscriptionGoogle_BalanceBelowMinimumReserve(t *testing.
rec := httptest.NewRecorder()
r.ServeHTTP(rec, req)
require.Equal(t, http.StatusOK, rec.Code)
}
func TestApiKeyAuthWithSubscriptionGoogle_RejectsExhaustedBalance(t *testing.T) {
gin.SetMode(gin.TestMode)
r := gin.New()
apiKeyService := newTestAPIKeyService(fakeAPIKeyRepo{
getByKey: func(ctx context.Context, key string) (*service.APIKey, error) {
return &service.APIKey{
ID: 1,
Key: key,
Status: service.StatusActive,
User: &service.User{
ID: 123,
Status: service.StatusActive,
Balance: 0,
},
}, nil
},
})
cfg := &config.Config{}
r.Use(APIKeyAuthWithSubscriptionGoogle(apiKeyService, nil, cfg))
r.GET("/v1beta/test", func(c *gin.Context) { c.JSON(200, gin.H{"ok": true}) })
req := httptest.NewRequest(http.MethodGet, "/v1beta/test", nil)
req.Header.Set("Authorization", "Bearer ok")
rec := httptest.NewRecorder()
r.ServeHTTP(rec, req)
require.Equal(t, http.StatusForbidden, rec.Code)
var resp googleErrorResponse
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp))
@@ -1000,7 +1000,7 @@ func TestAPIKeyAuthTouchesLastUsedInStandardMode(t *testing.T) {
require.Equal(t, 1, touchCalls)
}
func TestAPIKeyAuthRejectsBalanceBelowMinimumReserve(t *testing.T) {
func TestAPIKeyAuthAllowsBalanceBelowMinimumReserve(t *testing.T) {
gin.SetMode(gin.TestMode)
user := &service.User{
@@ -1039,6 +1039,49 @@ func TestAPIKeyAuthRejectsBalanceBelowMinimumReserve(t *testing.T) {
req.Header.Set("x-api-key", apiKey.Key)
router.ServeHTTP(w, req)
// 鉴权层保持历史语义:MinimumBalanceReserve 只用于 billing-cache 预检,
// 0 < balance < reserve 不得被鉴权中间件硬 403(存量部署静默行为变更)。
require.Equal(t, http.StatusOK, w.Code)
}
func TestAPIKeyAuthRejectsExhaustedBalance(t *testing.T) {
gin.SetMode(gin.TestMode)
user := &service.User{
ID: 10,
Role: service.RoleUser,
Status: service.StatusActive,
Balance: 0,
Concurrency: 3,
}
apiKey := &service.APIKey{
ID: 104,
UserID: user.ID,
Key: "held-balance-zero",
Status: service.StatusActive,
User: user,
}
apiKeyRepo := &stubApiKeyRepo{
getByKey: func(ctx context.Context, key string) (*service.APIKey, error) {
if key != apiKey.Key {
return nil, service.ErrAPIKeyNotFound
}
clone := *apiKey
userClone := *user
clone.User = &userClone
return &clone, nil
},
}
cfg := &config.Config{RunMode: config.RunModeStandard}
apiKeyService := service.NewAPIKeyService(apiKeyRepo, nil, nil, nil, nil, nil, cfg)
router := newAuthTestRouter(apiKeyService, nil, cfg)
w := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/t", nil)
req.Header.Set("x-api-key", apiKey.Key)
router.ServeHTTP(w, req)
require.Equal(t, http.StatusForbidden, w.Code)
requireAPIKeyAuthError(t, w, "INSUFFICIENT_BALANCE", "Insufficient account balance")
}