fix(v2): 超退守卫事务化 CreateRefundGuarded(锁单行+求和+插入单事务)+ sqlite DSN 补 _txlock=immediate

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 构造。
2026-07-10 17:08:25 +08:00
parent 338e28d386
commit aa14204640
4 changed files with 201 additions and 2 deletions
@@ -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 {
@@ -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 {
@@ -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)
+	}
+}
@@ -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{},