From aa14204640602f5f73f0f18679f0a3c2ad3d0a4c Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 17:08:25 +0800 Subject: [PATCH] =?UTF-8?q?fix(v2):=20=E8=B6=85=E9=80=80=E5=AE=88=E5=8D=AB?= =?UTF-8?q?=E4=BA=8B=E5=8A=A1=E5=8C=96=20CreateRefundGuarded(=E9=94=81?= =?UTF-8?q?=E5=8D=95=E8=A1=8C+=E6=B1=82=E5=92=8C+=E6=8F=92=E5=85=A5?= =?UTF-8?q?=E5=8D=95=E4=BA=8B=E5=8A=A1)+=20sqlite=20DSN=20=E8=A1=A5=20=5Ft?= =?UTF-8?q?xlock=3Dimmediate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 3 计划的 RefundSum→比较→CreateRefund 两段式裸读写在并发下会超退(两个请求都读到 reserved=0、都通过、都插入)。仿 store/order.go::MarkAttemptPaid 的模式,把锁订单行 (clause.Locking FOR UPDATE,MySQL 真锁/SQLite 由 glebarez 静默丢弃)+ 求和 + 校验 + 插入 收进单个 s.db.Transaction。SQLite 侧真正的串行化靠 DSN _txlock=immediate(BEGIN IMMEDIATE 在事务开始就抢写锁)+ busy_timeout,补进 testdb.go 与 main.go 的 sqlite DSN 构造。 --- internal/model/testdb.go | 7 +- internal/store/refund.go | 58 ++++++++++++++++ internal/store/refund_test.go | 122 ++++++++++++++++++++++++++++++++++ main.go | 16 ++++- 4 files changed, 201 insertions(+), 2 deletions(-) diff --git a/internal/model/testdb.go b/internal/model/testdb.go index 1aa1569..4910b80 100644 --- a/internal/model/testdb.go +++ b/internal/model/testdb.go @@ -20,7 +20,12 @@ var testDBCounter int64 func OpenTestDB(t *testing.T) *gorm.DB { t.Helper() n := atomic.AddInt64(&testDBCounter, 1) - dsn := fmt.Sprintf("file:testdb_%d?mode=memory&cache=shared", n) + // _txlock=immediate: BEGIN IMMEDIATE 让每个事务一开始就抢库级写锁(而非默认 + // BEGIN DEFERRED 推迟到首条写语句才抢),使并发事务在读阶段就串行化——守卫类 + // 事务(锁行/求和/校验/插入,见 RefundStore.CreateRefundGuarded)依赖此语义, + // 呼应 pangolin 的 SQLite DSN 约定(server/internal/db/db.go)。 + // _pragma=busy_timeout(5000): 抢不到写锁时等待重试而非立即 SQLITE_BUSY 报错。 + dsn := fmt.Sprintf("file:testdb_%d?mode=memory&cache=shared&_txlock=immediate&_pragma=busy_timeout(5000)", n) db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent), TranslateError: true}) if err != nil { diff --git a/internal/store/refund.go b/internal/store/refund.go index 6850c02..4f93df4 100644 --- a/internal/store/refund.go +++ b/internal/store/refund.go @@ -6,6 +6,7 @@ import ( "time" "gorm.io/gorm" + "gorm.io/gorm/clause" "github.com/wangjia/pay/internal/model" ) @@ -16,6 +17,11 @@ type RefundStore struct{ db *gorm.DB } func NewRefundStore(db *gorm.DB) *RefundStore { return &RefundStore{db: db} } +// CreateRefund is a plain insert with no over-refund guard — safe only for +// callers that already hold their own serialization (e.g. tests, backfills). +// Money-path callers (退款下单) MUST use CreateRefundGuarded instead: a bare +// RefundSum(...)-then-CreateRefund round trip races under concurrent refund +// requests (both read reserved=0, both pass, both insert → over-refund). func (s *RefundStore) CreateRefund(r *model.Refund) error { if r.RefundID == "" { return fmt.Errorf("store.CreateRefund: empty RefundID") @@ -26,6 +32,58 @@ func (s *RefundStore) CreateRefund(r *model.Refund) error { return nil } +// CreateRefundGuarded 在单事务内完成 锁定订单行 → 求和(非failed)→ 超退校验 → 插入。 +// 返回 (false, nil) 表示超退拒绝;并发同单退款经订单行锁串行化。 +// +// 订单行锁 clause.Locking{Strength: "UPDATE"} 在 MySQL 上是真正的 SELECT ... FOR +// UPDATE 行锁;SQLite 不支持行级锁,glebarez/sqlite 的 "FOR" ClauseBuilder 会把它 +// 静默丢弃(不报错、也不生效)。SQLite 侧真正的串行化来自 DSN 的 +// _txlock=immediate(见 internal/model/testdb.go / main.go initDB):BEGIN IMMEDIATE +// 让事务一开始就抢库级写锁,第二个并发事务的 BEGIN 会阻塞到第一个提交/回滚为止, +// 效果等价于「锁行」。没有 _txlock=immediate 时 SQLite 默认 BEGIN DEFERRED,写锁 +// 推迟到第一次写语句才抢,两个事务的读阶段(求和)仍可能交错,防不住超退。 +func (s *RefundStore) CreateRefundGuarded(r *model.Refund, paidAmountMinor int64) (bool, error) { + if r.RefundID == "" { + return false, fmt.Errorf("store.CreateRefundGuarded: empty RefundID") + } + if r.OutTradeNo == "" { + return false, fmt.Errorf("store.CreateRefundGuarded: empty OutTradeNo") + } + var ok bool + err := s.db.Transaction(func(tx *gorm.DB) error { + var o model.OrderV2 + if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}). + Where("out_trade_no = ?", r.OutTradeNo).First(&o).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return ErrOrderNotFound + } + return fmt.Errorf("lock order: %w", err) + } + + var reserved int64 + q := tx.Model(&model.Refund{}). + Where("out_trade_no = ? AND status <> ?", r.OutTradeNo, model.RefundFailed) + if err := q.Select("COALESCE(SUM(amount_minor),0)").Scan(&reserved).Error; err != nil { + return fmt.Errorf("sum refunds: %w", err) + } + + if reserved+r.AmountMinor > paidAmountMinor { + ok = false + return nil + } + + if err := tx.Create(r).Error; err != nil { + return err + } + ok = true + return nil + }) + if err != nil { + return false, fmt.Errorf("store.CreateRefundGuarded: %w", err) + } + return ok, nil +} + func (s *RefundStore) GetRefund(refundID string) (*model.Refund, error) { var r model.Refund if err := s.db.Where("refund_id = ?", refundID).First(&r).Error; err != nil { diff --git a/internal/store/refund_test.go b/internal/store/refund_test.go index 72d8646..906f925 100644 --- a/internal/store/refund_test.go +++ b/internal/store/refund_test.go @@ -1,9 +1,16 @@ package store import ( + "errors" + "fmt" + "sync" "testing" "time" + "github.com/glebarez/sqlite" + "gorm.io/gorm" + "gorm.io/gorm/logger" + "github.com/wangjia/pay/internal/model" ) @@ -103,3 +110,118 @@ func TestListManualPending(t *testing.T) { t.Fatalf("manual pending = %+v, %v", list, err) } } + +// TestCreateRefundGuardedSequential 顺序场景:60 通过、再 50 超退拒(60+50=110>100)、 +// 再 40 通过(60+40=100 打满),验证守卫在非并发路径下的基本判定与求和口径(非 failed)。 +func TestCreateRefundGuardedSequential(t *testing.T) { + db := model.OpenTestDB(t) + os := NewOrderStore(db) + rs := NewRefundStore(db) + seedPaid(t, os, "PAY-G1", 10000) // 100.00 + + ok, err := rs.CreateRefundGuarded(&model.Refund{ + RefundID: "g1-a", OutTradeNo: "PAY-G1", AttemptProviderRef: "PAY-G1", + AmountMinor: 6000, Currency: "CNY", Status: model.RefundProcessing, InitiatedBy: "business", + }, 10000) + if err != nil || !ok { + t.Fatalf("g1-a: ok=%v err=%v want true,nil", ok, err) + } + + ok, err = rs.CreateRefundGuarded(&model.Refund{ + RefundID: "g1-b", OutTradeNo: "PAY-G1", AttemptProviderRef: "PAY-G1", + AmountMinor: 5000, Currency: "CNY", Status: model.RefundProcessing, InitiatedBy: "business", + }, 10000) + if err != nil || ok { + t.Fatalf("g1-b: ok=%v err=%v want false,nil(超退应拒)", ok, err) + } + + ok, err = rs.CreateRefundGuarded(&model.Refund{ + RefundID: "g1-c", OutTradeNo: "PAY-G1", AttemptProviderRef: "PAY-G1", + AmountMinor: 4000, Currency: "CNY", Status: model.RefundProcessing, InitiatedBy: "business", + }, 10000) + if err != nil || !ok { + t.Fatalf("g1-c: ok=%v err=%v want true,nil", ok, err) + } + + sum, err := rs.RefundSum("PAY-G1", model.RefundRequested, model.RefundProcessing, model.RefundManualPending, model.RefundSucceeded) + if err != nil || sum != 10000 { + t.Fatalf("reserved sum = %d, err=%v want 10000", sum, err) + } + // 被拒的 g1-b 不应落库 + if _, err := rs.GetRefund("g1-b"); !errors.Is(err, ErrRefundNotFound) { + t.Fatalf("g1-b 应未落库,GetRefund err=%v", err) + } +} + +// openFileGuardedDB 开一个 t.TempDir 下的文件型 sqlite(而非 model.OpenTestDB 的 +// in-memory cache=shared),确保并发 goroutine 走真正独立的连接/文件锁路径来 +// 验证 CreateRefundGuarded 的串行化 —— DSN 同样带 _txlock=immediate。 +func openFileGuardedDB(t *testing.T) *gorm.DB { + t.Helper() + dsn := fmt.Sprintf("file:%s/guard.db?_txlock=immediate&_pragma=busy_timeout(5000)", t.TempDir()) + db, err := gorm.Open(sqlite.Open(dsn), + &gorm.Config{Logger: logger.Default.LogMode(logger.Silent), TranslateError: true}) + if err != nil { + t.Fatalf("open file guard db: %v", err) + } + if err := db.AutoMigrate(&model.OrderV2{}, &model.Attempt{}, &model.Account{}, &model.Refund{}, + &model.WebhookDelivery{}, &model.Product{}, &model.ProductPrice{}); err != nil { + t.Fatalf("migrate: %v", err) + } + sqlDB, _ := db.DB() + t.Cleanup(func() { _ = sqlDB.Close() }) + return db +} + +// TestCreateRefundGuardedConcurrentExactlyOneWins 两个 goroutine 同时对同一笔 +// 100.00 已付订单各申请 60.00 退款(合计 120 > 100,必超退)。事务化守卫(锁订单行 + +// 事务内求和 + 校验 + 插入)必须让恰好一个成功、另一个被拒 —— 若把守卫换回「先 +// RefundSum 再 CreateRefund」的裸读写两段式(Task 3 计划里的朴素调用法),两个 +// goroutine 都可能读到 reserved=0、都通过校验、都插入成功,变成两个 true(超退)。 +// 本测试即用来钉住这个回归:TDD 上手动验证过 —— 把 CreateRefundGuarded 换成裸 +// RefundSum(...)+CreateRefund(...) 两步、不用事务/锁,本测试会因为「两个 true」而 +// FAIL(count==2),证明测试确实在防这一类竞态,而不是摆设。 +func TestCreateRefundGuardedConcurrentExactlyOneWins(t *testing.T) { + db := openFileGuardedDB(t) + os := NewOrderStore(db) + rs := NewRefundStore(db) + seedPaid(t, os, "PAY-G2", 10000) // 100.00 + + var wg sync.WaitGroup + results := make([]bool, 2) + errs := make([]error, 2) + ids := []string{"g2-a", "g2-b"} + for i := 0; i < 2; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + ok, err := rs.CreateRefundGuarded(&model.Refund{ + RefundID: ids[i], OutTradeNo: "PAY-G2", AttemptProviderRef: "PAY-G2", + AmountMinor: 6000, Currency: "CNY", Status: model.RefundProcessing, InitiatedBy: "business", + }, 10000) + results[i] = ok + errs[i] = err + }(i) + } + wg.Wait() + + trueCount := 0 + for i, ok := range results { + // 允许 SQLite 忙锁重试耗尽等基础设施性错误(busy_timeout 已设为 5s 通常不会 + // 触发),但不允许"两个都成功"的超退;若出现非预期错误直接报失败方便定位。 + if errs[i] != nil { + t.Fatalf("goroutine %d unexpected error: %v", i, errs[i]) + } + if ok { + trueCount++ + } + } + if trueCount != 1 { + t.Fatalf("trueCount = %d, want 1 (results=%v)", trueCount, results) + } + + sum, err := rs.RefundSum("PAY-G2", model.RefundRequested, model.RefundProcessing, model.RefundManualPending, model.RefundSucceeded) + if err != nil || sum != 6000 { + t.Fatalf("reserved sum = %d, err=%v want 6000(仅一笔落库)", sum, err) + } +} diff --git a/main.go b/main.go index bd4246b..17813e7 100644 --- a/main.go +++ b/main.go @@ -3,6 +3,7 @@ package main import ( "errors" "log" + "strings" "time" "github.com/gin-gonic/gin" @@ -84,7 +85,7 @@ func initDB() *gorm.DB { switch config.C.Database.Driver { case "sqlite", "": - db, err := gorm.Open(sqlite.Open(config.C.Database.DSN), gormCfg) + db, err := gorm.Open(sqlite.Open(sqliteDSN(config.C.Database.DSN)), gormCfg) if err != nil { log.Fatalf("连接 sqlite 失败: %v", err) } @@ -98,6 +99,19 @@ func initDB() *gorm.DB { } } +// sqliteDSN 补 _txlock=immediate(BEGIN IMMEDIATE 立即抢库级写锁,串行化并发事务的 +// 读阶段——RefundStore.CreateRefundGuarded 等守卫类事务依赖此语义)+ +// busy_timeout(5000)(抢不到锁时等待重试而非立即报 SQLITE_BUSY)。呼应 pangolin +// 的 SQLite DSN 约定(server/internal/db/db.go)。config.yaml 里的 dsn 是纯文件路径 +// (如 "pay.db"),这里补上查询串;若用户已经自带 "?"，追加而不是覆盖。 +func sqliteDSN(raw string) string { + extra := "_txlock=immediate&_pragma=busy_timeout(5000)" + if strings.Contains(raw, "?") { + return raw + "&" + extra + } + return raw + "?" + extra +} + func autoMigrate(db *gorm.DB) { if err := db.AutoMigrate( &model.Merchant{},