From 4cc6ff307e7cca8cb37b3ec7426afc6cf47d163d Mon Sep 17 00:00:00 2001 From: martin <406615373@qq.com> Date: Sun, 26 Jul 2026 12:41:28 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20P1=20=E4=BA=8B?= =?UTF-8?q?=E5=8A=A1=E8=BE=B9=E7=95=8C=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P1-T1 SetTenantGroups/SetGroupModels 删-插包进事务，避免插入失败清空关联 - P1-T2 ApiKeyUpdate 主表与 scope 更新同事务，缓存失效移到提交后 - P1-T3 InvalidateWallet 移到 FulfillOrder 事务提交后，避免缓存污染窗口 - P1-T4 handleDeletionRequest 跨 4 表删改包进事务，杜绝中间态 - P1-T5 subscribePlanTx 加事务级 advisory lock，串行化同租户订阅 --- internal/logic/admin/model_group.go | 86 +++++++++++++++++------------ internal/logic/payment/fulfill.go | 37 ++++++++++++- internal/logic/task/data_tasks.go | 58 +++++++++++-------- internal/logic/tenant/api_key.go | 36 +++++++----- 代码分析结果.md | 15 +++-- 5 files changed, 152 insertions(+), 80 deletions(-) diff --git a/internal/logic/admin/model_group.go b/internal/logic/admin/model_group.go index afb2874..0c746b3 100644 --- a/internal/logic/admin/model_group.go +++ b/internal/logic/admin/model_group.go @@ -4,6 +4,7 @@ import ( "context" "fmt" + "github.com/gogf/gf/v2/database/gdb" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" v1 "github.com/qianfree/team-api/api/admin/v1" @@ -256,27 +257,35 @@ func (s *sAdmin) SetGroupModels(ctx context.Context, req *v1.GroupModelsSetReq) return nil, common.NewBusinessError(consts.CodeModelGroupNotFound, consts.MsgModelGroupNotFound) } - // 删除旧关联 - _, err := dao.MdlGroupModels.Ctx(ctx).Where("group_id", req.ID).Delete() + // 删-插必须在同一事务内：插入失败（DB 抖动/唯一键冲突）若与删除分离，会清空分组关联， + // 影响下游按分组计算的租户可用模型与定价。事务采用 ctx 传播式：闭包内统一 dao.Xxx.Ctx(ctx)。 + err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { + // 删除旧关联 + _, err := dao.MdlGroupModels.Ctx(ctx).Where("group_id", req.ID).Delete() + if err != nil { + return err + } + + // 批量插入新关联 + if len(req.ModelIds) > 0 { + insertData := make([]do.MdlGroupModels, 0, len(req.ModelIds)) + for _, modelID := range req.ModelIds { + insertData = append(insertData, do.MdlGroupModels{ + GroupId: req.ID, + ModelId: modelID, + }) + } + _, err = dao.MdlGroupModels.Ctx(ctx).Batch(len(insertData)).Insert(insertData) + if err != nil { + return err + } + } + return nil + }) if err != nil { return nil, err } - // 批量插入新关联 - if len(req.ModelIds) > 0 { - insertData := make([]do.MdlGroupModels, 0, len(req.ModelIds)) - for _, modelID := range req.ModelIds { - insertData = append(insertData, do.MdlGroupModels{ - GroupId: req.ID, - ModelId: modelID, - }) - } - _, err = dao.MdlGroupModels.Ctx(ctx).Batch(len(insertData)).Insert(insertData) - if err != nil { - return nil, err - } - } - invalidateTenantsInGroup(ctx, req.ID) return nil, nil @@ -321,27 +330,36 @@ func (s *sAdmin) ListTenantGroups(ctx context.Context, req *v1.TenantGroupsListR // SetTenantGroups 设置租户关联的分组（全量替换） func (s *sAdmin) SetTenantGroups(ctx context.Context, req *v1.TenantGroupsSetReq) (*v1.TenantGroupsSetRes, error) { - // 删除旧关联 - _, err := dao.MdlTenantGroups.Ctx(ctx).Where("tenant_id", req.TenantID).Delete() + // 删-插必须在同一事务内：插入失败（DB 抖动/唯一键冲突）若与删除分离，会清空租户模型分组， + // 影响下游按分组计算的租户可用模型与定价。事务采用 ctx 传播式：闭包内统一 dao.Xxx.Ctx(ctx)， + // 参照 tenant/member_model_scope.go 的正确写法。 + err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { + // 删除旧关联 + _, err := dao.MdlTenantGroups.Ctx(ctx).Where("tenant_id", req.TenantID).Delete() + if err != nil { + return err + } + + // 批量插入新关联 + if len(req.GroupIds) > 0 { + insertData := make([]do.MdlTenantGroups, 0, len(req.GroupIds)) + for _, groupID := range req.GroupIds { + insertData = append(insertData, do.MdlTenantGroups{ + TenantId: req.TenantID, + GroupId: groupID, + }) + } + _, err = dao.MdlTenantGroups.Ctx(ctx).Batch(len(insertData)).Insert(insertData) + if err != nil { + return err + } + } + return nil + }) if err != nil { return nil, err } - // 批量插入新关联 - if len(req.GroupIds) > 0 { - insertData := make([]do.MdlTenantGroups, 0, len(req.GroupIds)) - for _, groupID := range req.GroupIds { - insertData = append(insertData, do.MdlTenantGroups{ - TenantId: req.TenantID, - GroupId: groupID, - }) - } - _, err = dao.MdlTenantGroups.Ctx(ctx).Batch(len(insertData)).Insert(insertData) - if err != nil { - return nil, err - } - } - invalidateTenantGroupCache(ctx, req.TenantID) billing.ClearTenantPriceCache(ctx, req.TenantID) diff --git a/internal/logic/payment/fulfill.go b/internal/logic/payment/fulfill.go index 231e55a..59dc6e9 100644 --- a/internal/logic/payment/fulfill.go +++ b/internal/logic/payment/fulfill.go @@ -18,7 +18,12 @@ import ( // FulfillOrder 履约订单（事务内完成：履约+更新订单状态） func FulfillOrder(ctx context.Context, orderID int64) error { - return g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { + // walletTouched 标记本次履约是否变更了钱包余额；钱包缓存失效必须在事务提交后执行， + // 否则并发 GetWallet 可能在提交前读到旧 DB 值并回写 Redis，污染缓存。 + var walletTouchedTenantID int64 + var walletTouched bool + + err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { var order *struct { TenantID int64 `json:"tenant_id"` UserID int64 `json:"user_id"` @@ -65,6 +70,8 @@ func FulfillOrder(ctx context.Context, orderID int64) error { _ = billing.CheckAndUpgradeLevel(ctx, order.TenantID) // 充值后检查是否需要重置低余额预警标记 billing.ResetLowBalanceNotified(ctx, order.TenantID) + walletTouchedTenantID = order.TenantID + walletTouched = true default: return gerror.Newf("unsupported order type for fulfillment: %s", order.OrderType) @@ -90,10 +97,33 @@ func FulfillOrder(ctx context.Context, orderID int64) error { } return nil }) + if err != nil { + return err + } + + // 事务提交后再清除钱包两级缓存（进程内 walletCache + Redis），避免充值后 GetWallet 在 300s 内仍返回旧余额。 + // 放在提交后执行，避免「DEL → 并发 GetWallet 读到旧 DB 值回写」的缓存污染窗口。 + if walletTouched { + billing.InvalidateWallet(ctx, walletTouchedTenantID) + } + return nil } +// planSubscribeLockNS 是套餐订阅串行化 advisory lock 的命名空间前缀。 +// 采用单 key 形式 pg_advisory_xact_lock(bigint)，key = NS + tenantID： +// NS 取高位常量段，与其它业务模块（直接以主键为 key）的 advisory lock 隔离，避免冲突。 +const planSubscribeLockNS int64 = 9_001_000_000_000_000 + // subscribePlanTx 在事务内订阅套餐（依赖调用方传入携带事务的 ctx，内部统一用 dao.Xxx.Ctx(ctx) 传播） func subscribePlanTx(ctx context.Context, tenantID int64, planID int64, months int, autoRenew bool) error { + // 串行化同一租户的套餐订阅：两个并发履约（兑换码 + 自动续费、双支付回调等）可能同时读到 + // active 订阅并各自「置 expired + INSERT 新 active」，最终留下两条 active 订阅。 + // pg_advisory_xact_lock 随事务提交/回滚自动释放，且不占用任何行锁，无死锁风险； + // 后到事务阻塞至前者提交（旧 active 已 expired）后再进入临界区，据此跳过重复订阅。 + if _, err := g.DB().Ctx(ctx).Exec(ctx, "SELECT pg_advisory_xact_lock(?)", planSubscribeLockNS+tenantID); err != nil { + return gerror.Wrapf(err, "acquire subscribe advisory lock for tenant %d", tenantID) + } + var plan *struct { MonthlyPrice float64 `json:"monthly_price"` YearlyPrice float64 `json:"yearly_price"` @@ -203,7 +233,8 @@ func creditWalletTx(ctx context.Context, tenantID int64, amount decimal.Decimal, return err } - // 清除钱包两级缓存（进程内 walletCache + Redis），避免充值后 GetWallet 在 300s 内仍返回旧余额 - billing.InvalidateWallet(ctx, tenantID) + // 注意：钱包缓存失效（billing.InvalidateWallet）不在事务内调用——事务尚未提交时 DEL Redis， + // 并发 GetWallet 可能从 DB 读到旧余额并回写 Redis，导致缓存被旧值重新污染。 + // 失效由调用方 FulfillOrder 在事务提交后统一执行。 return nil } diff --git a/internal/logic/task/data_tasks.go b/internal/logic/task/data_tasks.go index 83ac0c5..0d5d0e9 100644 --- a/internal/logic/task/data_tasks.go +++ b/internal/logic/task/data_tasks.go @@ -7,6 +7,7 @@ import ( "fmt" "time" + "github.com/gogf/gf/v2/database/gdb" "github.com/gogf/gf/v2/frame/g" "github.com/qianfree/team-api/internal/dao" @@ -153,31 +154,40 @@ func handleDeletionRequest(ctx context.Context, payload json.RawMessage) (any, e LogInfo(ctx, 0, fmt.Sprintf("开始处理租户 %d 的数据删除请求", p.TenantID)) - _, err := dao.TntUsers.Ctx(ctx). - Where("tenant_id", p.TenantID). - Data(do.TntUsers{ - DisplayName: "[deleted]", - Email: fmt.Sprintf("deleted_%d@deleted.local", p.TenantID), - }).Update() + // 跨 4 张表（tnt_users / api_keys / aud_sensitive_access_logs / tnt_tenants）的删除与更新必须 + // 在同一事务内完成，否则中途失败会留下部分应用的中间态（如 API Key 已禁用但租户未标记 terminated）。 + // 事务采用 ctx 传播式：闭包内统一 dao.Xxx.Ctx(ctx)，事务由 ctx 自动挂载。 + err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { + _, err := dao.TntUsers.Ctx(ctx). + Where("tenant_id", p.TenantID). + Data(do.TntUsers{ + DisplayName: "[deleted]", + Email: fmt.Sprintf("deleted_%d@deleted.local", p.TenantID), + }).Update() + if err != nil { + LogError(ctx, 0, fmt.Sprintf("匿名化用户数据失败: %v", err)) + return fmt.Errorf("anonymize users: %w", err) + } + + if _, err := dao.ApiKeys.Ctx(ctx). + Where("tenant_id", p.TenantID). + Data(do.ApiKeys{Status: "disabled"}).Update(); err != nil { + return fmt.Errorf("disable api keys: %w", err) + } + + if _, err := common.AuditModelCtx(ctx, "aud_sensitive_access_logs").Where("tenant_id", p.TenantID).Delete(); err != nil { + return fmt.Errorf("delete sensitive access logs: %w", err) + } + + if _, err := dao.TntTenants.Ctx(ctx). + Where("id", p.TenantID). + Data(do.TntTenants{Status: "terminated"}).Update(); err != nil { + return fmt.Errorf("terminate tenant: %w", err) + } + return nil + }) if err != nil { - LogError(ctx, 0, fmt.Sprintf("匿名化用户数据失败: %v", err)) - return nil, fmt.Errorf("anonymize users: %w", err) - } - - if _, err := dao.ApiKeys.Ctx(ctx). - Where("tenant_id", p.TenantID). - Data(do.ApiKeys{Status: "disabled"}).Update(); err != nil { - return nil, fmt.Errorf("disable api keys: %w", err) - } - - if _, err := common.AuditModelCtx(ctx, "aud_sensitive_access_logs").Where("tenant_id", p.TenantID).Delete(); err != nil { - return nil, fmt.Errorf("delete sensitive access logs: %w", err) - } - - if _, err := dao.TntTenants.Ctx(ctx). - Where("id", p.TenantID). - Data(do.TntTenants{Status: "terminated"}).Update(); err != nil { - return nil, fmt.Errorf("terminate tenant: %w", err) + return nil, err } proof := map[string]any{ diff --git a/internal/logic/tenant/api_key.go b/internal/logic/tenant/api_key.go index acfc625..57d4190 100644 --- a/internal/logic/tenant/api_key.go +++ b/internal/logic/tenant/api_key.go @@ -443,23 +443,31 @@ func (s *sTenant) ApiKeyUpdate(ctx context.Context, req *v1.TenantApiKeyUpdateRe hasUpdate = true } - if hasUpdate { - _, err = dao.ApiKeys.Ctx(ctx). - Where("id", keyID). - Where("tenant_id", tenantID). - Data(data). - Update() - if err != nil { - return nil, err + // 主表更新与模型范围更新必须在同一事务内：前者成功后者失败会留下中间态（如 Key 已改名但 + // scope 仍为旧值）。外层事务包住两步；updateApiKeyModelScopes 在携带事务的 ctx 下会被 + // GoFrame 视为嵌套事务（savepoint），与主表更新共享同一事务。缓存失效放在事务提交后执行。 + err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { + if hasUpdate { + _, err = dao.ApiKeys.Ctx(ctx). + Where("id", keyID). + Where("tenant_id", tenantID). + Data(data). + Update() + if err != nil { + return err + } } - } - // 更新模型范围（事务内先删后插） - if req.ModelNames != nil { - err = updateApiKeyModelScopes(ctx, keyID, req.ModelNames) - if err != nil { - return nil, err + // 更新模型范围（事务内先删后插，作为外层事务的 savepoint） + if req.ModelNames != nil { + if err = updateApiKeyModelScopes(ctx, keyID, req.ModelNames); err != nil { + return err + } } + return nil + }) + if err != nil { + return nil, err } if hasUpdate || req.ModelNames != nil { diff --git a/代码分析结果.md b/代码分析结果.md index a4144a2..b26c327 100644 --- a/代码分析结果.md +++ b/代码分析结果.md @@ -205,37 +205,42 @@ ### 🔄 P1-Transaction　事务边界类 -#### - [ ] P1-T1　SetTenantGroups 删-插无事务 +#### - [x] P1-T1　SetTenantGroups 删-插无事务 - **位置**：`internal/logic/admin/model_group.go:323-349` - **问题描述**：先 `Delete()` 旧关联，再 `Batch().Insert()` 新关联，两步不在同一事务。 - **影响**：插入失败（DB 抖动/唯一键冲突）会清空租户模型分组，影响下游定价。 - **修复建议**：包 `g.DB().Transaction`，参照项目正确写法 `tenant/member_model_scope.go:88`。 +　**〔已修复〕** `SetTenantGroups` 的 Delete+BatchInsert 包进 `g.DB().Transaction`（ctx 传播式，闭包内统一 `dao.Xxx.Ctx(ctx)`）。同文件 `SetGroupModels` 存在相同删-插分离 bug，一并包进事务。 -#### - [ ] P1-T2　ApiKeyUpdate 主表与 scope 更新分离 +#### - [x] P1-T2　ApiKeyUpdate 主表与 scope 更新分离 - **位置**：`internal/logic/tenant/api_key.go:457-463` - **问题描述**：Key 主表更新与 `updateApiKeyModelScopes` 两步分离，前者成功后者失败留中间态。 - **修复建议**：外层事务包住两步。 +　**〔已修复〕** ApiKeyUpdate 主表 `Update` 与 `updateApiKeyModelScopes` 用外层 `g.DB().Transaction` 包住；后者在携带事务的 ctx 下被 GoFrame 视为嵌套事务（savepoint），与主表更新共享同一事务，任一失败整体回滚。`relay.InvalidateApiKey` 缓存失效移到事务提交后执行。 -#### - [ ] P1-T3　InvalidateWallet 在事务提交前调用 +#### - [x] P1-T3　InvalidateWallet 在事务提交前调用 - **位置**：`internal/logic/payment/fulfill.go:207` - **问题描述**：事务闭包内调 `billing.InvalidateWallet`（Redis DEL），但事务尚未提交。并发 `GetWallet` 可能从 DB 读到旧值回写 Redis。 - **修复建议**：把 `InvalidateWallet` 移到事务外（`FulfillOrder` 返回后调用）。 +　**〔已修复〕** 移除 `creditWalletTx` 内的 `billing.InvalidateWallet`；`FulfillOrder` 改为先记录 `walletTouched` 标记，事务提交成功后再调 `billing.InvalidateWallet`（与 `redemption.go:152` 的「事务提交后清缓存」模式一致）。 -#### - [ ] P1-T4　handleDeletionRequest 跨表无事务 +#### - [x] P1-T4　handleDeletionRequest 跨表无事务 - **位置**：`internal/logic/task/data_tasks.go:111-155` - **问题描述**：跨 `tnt_users`、`api_keys`、`aud_sensitive_access_logs`、`tnt_tenants` 4 张表删除/更新无事务。 - **影响**：中途失败留下部分应用的中间态（如 API Key 已禁用但租户未标记 terminated）。 - **修复建议**：`g.DB().Transaction` 包裹。 +　**〔已修复〕** 4 张表的匿名化/禁用/删除/终止包进 `g.DB().Transaction`，任一步失败整体回滚，杜绝部分应用的中间态。 -#### - [ ] P1-T5　subscribePlanTx 先取消当前订阅无并发保护 +#### - [x] P1-T5　subscribePlanTx 先取消当前订阅无并发保护 - **位置**：`internal/logic/payment/fulfill.go:120-130` - **问题描述**：把当前 active 订阅置 expired 再 INSERT 新的，两个并发履约可能都读到 active → 两条 active 订阅。 - **修复建议**：INSERT 后用唯一约束（tenant_id + status='active'）防重，或事务内 advisory lock。 +　**〔已修复〕** 采用事务内 advisory lock：`subscribePlanTx` 入口 `SELECT pg_advisory_xact_lock(NS+tenantID)`（`planSubscribeLockNS` 高位命名空间段，避免与其他模块冲突）。事务级锁随提交/回滚自动释放，不占用行锁无死锁风险；后到事务阻塞至前者提交（旧 active 已 expired）后进入临界区，串行化同租户订阅。 ---