okooooooo/dujiao-next
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修复订单风控死锁bug issues#271

Browse Source
This commit is contained in:
dujiao-next committed 2026-08-05 17:33:59 +08:00
1 parent 35c5fece51
commit 041c0e4037
5 files changed
+168 -15
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/modules/order/application/order_service.go
+4 -1
View File
@@ -282,6 +282,7 @@ type orderCreateParams struct {
ClientIP string
RiskIP string
RiskPaymentExpireMinutes int
RiskCheckResult orderriskcontract.CheckResult
IsGuest bool
ManualFormData map[string]jsonmap.JSON
SkipManualFormCheck bool
@@ -542,7 +543,7 @@ func (s *OrderService) createOrder(input orderCreateParams) (*orderdomain.Order,
orderStore := tx.Orders()
productSKURepo := tx.ProductSKUs()
if s.riskControlSvc != nil && !input.SkipRiskControl {
if err := s.riskControlSvc.CheckPendingOrderAllowed(buildRiskCheckInput(input, false), orderStore); err != nil {
if err := s.riskControlSvc.CheckPendingOrderAllowed(buildRiskCheckInput(input, false), input.RiskCheckResult, orderStore); err != nil {
return err
}
}
@@ -728,6 +729,7 @@ func (s *OrderService) checkOrderRisk(input *orderCreateParams, consumeRateLimit
input.ClientIP = strings.TrimSpace(input.ClientIP)
input.RiskIP = orderriskcontract.NormalizeRiskIP(input.ClientIP)
input.RiskPaymentExpireMinutes = 0
input.RiskCheckResult = orderriskcontract.CheckResult{}
if s.riskControlSvc == nil || input.SkipRiskControl {
return nil
}
@@ -737,6 +739,7 @@ func (s *OrderService) checkOrderRisk(input *orderCreateParams, consumeRateLimit
}
input.RiskIP = result.RiskIP
input.RiskPaymentExpireMinutes = result.PaymentExpireMinutes
input.RiskCheckResult = result
return nil
}
internal/modules/orderrisk/application/service.go
+17 -9
View File
@@ -40,17 +40,24 @@ func NewService(options Options) *Service {
return &Service{settings: options.Settings, rateLimiter: options.RateLimiter}
}
// CheckOrderAllowed 执行不依赖订单事务的风控，并返回后续事务应复用的规范化 IP。
// CheckOrderAllowed 在订单事务外读取配置并执行前置风控，返回后续事务应复用的配置与规范化 IP。
func (s *Service) CheckOrderAllowed(input orderriskcontract.CheckInput) (orderriskcontract.CheckResult, error) {
result := orderriskcontract.CheckResult{RiskIP: orderriskcontract.NormalizeRiskIP(input.ClientIP)}
result := orderriskcontract.CheckResult{
RiskIP: orderriskcontract.NormalizeRiskIP(input.ClientIP),
ConfigSnapshot: settingssecurity.DefaultOrderRiskControlConfig(),
}
if s == nil || s.settings == nil {
return result, nil
}
cfg, err := s.settings.GetOrderRiskControlConfig()
if err != nil {
logger.Warnw("risk_control_get_config_error", "error", err)
if input.ConsumeRateLimit {
return result, err
}
return result, nil
}
result.ConfigSnapshot = cfg
if !cfg.Enabled {
return result, nil
}
@@ -135,19 +142,20 @@ func aggregateProductQuantities(items []orderriskcontract.OrderItem) map[uint]in
}
// CheckPendingOrderAllowed 必须在订单事务内、创建父订单和锁库存之前调用。
func (s *Service) CheckPendingOrderAllowed(input orderriskcontract.CheckInput, gate orderriskcontract.PendingOrderGate) error {
if s == nil || s.settings == nil || gate == nil {
// prepared 来自事务外的 CheckOrderAllowed；事务内禁止再次读取独立设置仓储。
func (s *Service) CheckPendingOrderAllowed(input orderriskcontract.CheckInput, prepared orderriskcontract.CheckResult, gate orderriskcontract.PendingOrderGate) error {
if s == nil || gate == nil {
return nil
}
cfg, err := s.settings.GetOrderRiskControlConfig()
if err != nil {
return err
}
cfg := prepared.ConfigSnapshot
if !cfg.Enabled {
return nil
}
if input.RiskIP == "" {
input.RiskIP = orderriskcontract.NormalizeRiskIP(input.ClientIP)
input.RiskIP = prepared.RiskIP
if input.RiskIP == "" {
input.RiskIP = orderriskcontract.NormalizeRiskIP(input.ClientIP)
}
}
if input.IsGuest {
return checkGuestPendingOrder(input, cfg.Guest, gate)
internal/modules/orderrisk/application/service_test.go
+142 -3
View File
@@ -2,11 +2,23 @@ package application
import (
"errors"
"path/filepath"
"reflect"
"testing"
"time"
"github.com/dujiao-next/internal/constants"
ordercontract "github.com/dujiao-next/internal/modules/order/contract"
orderdomain "github.com/dujiao-next/internal/modules/order/domain"
orderstore "github.com/dujiao-next/internal/modules/order/infrastructure/gormstore"
orderriskcontract "github.com/dujiao-next/internal/modules/orderrisk/contract"
orderriskdomain "github.com/dujiao-next/internal/modules/orderrisk/domain"
settingsapp "github.com/dujiao-next/internal/modules/settings/application"
settingsstore "github.com/dujiao-next/internal/modules/settings/infrastructure/gormstore"
settingssecurity "github.com/dujiao-next/internal/modules/settings/schema/security"
"github.com/glebarez/sqlite"
"gorm.io/gorm"
)
type settingReaderStub struct {
@@ -18,6 +30,17 @@ func (s settingReaderStub) GetOrderRiskControlConfig() (settingssecurity.OrderRi
return s.config, s.err
}
type countingSettingReader struct {
config settingssecurity.OrderRiskControlConfig
err error
calls int
}
func (s *countingSettingReader) GetOrderRiskControlConfig() (settingssecurity.OrderRiskControlConfig, error) {
s.calls++
return s.config, s.err
}
type rateLimiterStub struct {
calls int
input orderriskcontract.CheckInput
@@ -79,6 +102,118 @@ func TestCheckOrderAllowed_DisabledByDefault(t *testing.T) {
}
}
func TestCheckPendingOrderAllowedDoesNotReadSettingsInsideSingleConnectionTransaction(t *testing.T) {
dsn := filepath.Join(t.TempDir(), "risk-single-connection.db")
db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{})
if err != nil {
t.Fatalf("open sqlite: %v", err)
}
if err := db.AutoMigrate(
&settingsstore.SettingRecord{},
&orderdomain.Order{},
&orderdomain.OrderItem{},
&orderriskdomain.LockKey{},
); err != nil {
t.Fatalf("migrate risk tables: %v", err)
}
cfg := testConfig()
if err := db.Create(&settingsstore.SettingRecord{
Key: constants.SettingKeyOrderRiskControlConfig,
ValueJSON: settingssecurity.EncodeOrderRiskControlConfig(cfg),
}).Error; err != nil {
t.Fatalf("seed enabled risk config: %v", err)
}
sqlDB, err := db.DB()
if err != nil {
t.Fatalf("get sql db: %v", err)
}
t.Cleanup(func() { _ = sqlDB.Close() })
sqlDB.SetMaxOpenConns(1)
sqlDB.SetMaxIdleConns(1)
settings := settingsapp.NewService(settingsstore.New(db))
svc := NewService(Options{Settings: settings})
orders := orderstore.New(db, "risk-single-connection-secret-with-32-bytes")
input := orderriskcontract.CheckInput{
IsGuest: true,
ClientIP: "1.2.3.4",
Items: []orderriskcontract.OrderItem{{ProductID: 7, Quantity: 1}},
}
prepared, err := svc.CheckOrderAllowed(input)
if err != nil {
t.Fatalf("prepare risk check: %v", err)
}
if !prepared.ConfigSnapshot.Enabled {
t.Fatal("expected enabled risk config snapshot")
}
done := make(chan error, 1)
go func() {
done <- orders.WithinTransaction(func(tx ordercontract.Transaction) error {
return svc.CheckPendingOrderAllowed(input, prepared, tx.Orders())
})
}()
select {
case err := <-done:
if err != nil {
t.Fatalf("pending risk check: %v", err)
}
case <-time.After(2 * time.Second):
stats := sqlDB.Stats()
// 释放旧实现中等待第二条连接的 goroutine，避免失败测试泄漏。
sqlDB.SetMaxOpenConns(2)
select {
case <-done:
case <-time.After(time.Second):
}
t.Fatalf("pending risk check blocked inside transaction: in_use=%d open=%d wait_count=%d", stats.InUse, stats.OpenConnections, stats.WaitCount)
}
}
func TestPendingRiskCheckReusesPreparedConfigSnapshot(t *testing.T) {
cfg := testConfig()
settings := &countingSettingReader{config: cfg}
svc := NewService(Options{Settings: settings})
input := orderriskcontract.CheckInput{IsGuest: true, ClientIP: "1.2.3.4"}
prepared, err := svc.CheckOrderAllowed(input)
if err != nil {
t.Fatalf("prepare risk check: %v", err)
}
if err := svc.CheckPendingOrderAllowed(input, prepared, &pendingGateStub{}); err != nil {
t.Fatalf("pending risk check: %v", err)
}
if settings.calls != 1 {
t.Fatalf("settings reads=%d, want exactly one before the transaction", settings.calls)
}
}
func TestCheckOrderAllowed_SettingsReadFailureFailsClosedOnlyForCreate(t *testing.T) {
settingsErr := errors.New("settings unavailable")
svc := NewService(Options{Settings: settingReaderStub{err: settingsErr}})
preview, err := svc.CheckOrderAllowed(orderriskcontract.CheckInput{
IsGuest: true,
ClientIP: "1.2.3.4",
})
if err != nil {
t.Fatalf("preview-style check should preserve fail-open behavior, got %v", err)
}
if preview.ConfigSnapshot.Enabled {
t.Fatal("preview fallback config must keep risk control disabled")
}
_, err = svc.CheckOrderAllowed(orderriskcontract.CheckInput{
IsGuest: true,
ClientIP: "1.2.3.4",
ConsumeRateLimit: true,
})
if !errors.Is(err, settingsErr) {
t.Fatalf("create-style check must fail closed, got %v", err)
}
}
func TestCheckOrderAllowed_IPBlacklistCanonicalizesAddress(t *testing.T) {
cfg := testConfig()
cfg.Common.IPBlacklist = []string{"1.2.3.4", "10.0.0.0/8"}
@@ -161,7 +296,7 @@ func TestCheckPendingOrderAllowed_GuestLocksIPAndCountsOrders(t *testing.T) {
IsGuest: true,
RiskIP: "1.2.3.4",
Items: []orderriskcontract.OrderItem{{ProductID: 1, Quantity: 1}},
}, gate)
}, orderriskcontract.CheckResult{ConfigSnapshot: cfg}, gate)
if !errors.Is(err, orderriskcontract.ErrTooManyPendingOrders) {
t.Fatalf("expected guest pending order limit, got %v", err)
}
@@ -184,7 +319,7 @@ func TestCheckPendingOrderAllowed_GuestCountsPendingProductQuantity(t *testing.T
{ProductID: 7, Quantity: 1},
{ProductID: 7, Quantity: 1},
},
}, gate)
}, orderriskcontract.CheckResult{ConfigSnapshot: cfg}, gate)
if !errors.Is(err, orderriskcontract.ErrPendingProductQuantityLimit) {
t.Fatalf("expected pending product quantity limit, got %v", err)
}
@@ -197,7 +332,11 @@ func TestCheckPendingOrderAllowed_MemberUsesUserAndOptionalIP(t *testing.T) {
gate := &pendingGateStub{pendingByUser: 4, pendingMemberIP: 3}
svc := NewService(Options{Settings: settingReaderStub{config: cfg}})
err := svc.CheckPendingOrderAllowed(orderriskcontract.CheckInput{UserID: 9, RiskIP: "1.2.3.4"}, gate)
err := svc.CheckPendingOrderAllowed(
orderriskcontract.CheckInput{UserID: 9, RiskIP: "1.2.3.4"},
orderriskcontract.CheckResult{ConfigSnapshot: cfg},
gate,
)
if !errors.Is(err, orderriskcontract.ErrTooManyPendingOrders) {
t.Fatalf("expected member IP pending limit, got %v", err)
}
internal/modules/orderrisk/contract/ports.go
+1 -1
View File
@@ -23,5 +23,5 @@ type RateLimiter interface {
// Controller 是订单上下文调用风控所需的用例端口。
type Controller interface {
CheckOrderAllowed(input CheckInput) (CheckResult, error)
CheckPendingOrderAllowed(input CheckInput, gate PendingOrderGate) error
CheckPendingOrderAllowed(input CheckInput, prepared CheckResult, gate PendingOrderGate) error
}
internal/modules/orderrisk/contract/types.go
+4 -1
View File
@@ -3,6 +3,8 @@ package contract
import (
"net"
"strings"
settingssecurity "github.com/dujiao-next/internal/modules/settings/schema/security"
)
// OrderItem 是风控所需的最小订单项快照。
@@ -22,10 +24,11 @@ type CheckInput struct {
Items []OrderItem
}
// CheckResult 返回创建订单后续步骤需要复用的规范化风控上下文。
// CheckResult 返回创建订单后续步骤需要复用的不可变风控上下文。
type CheckResult struct {
RiskIP string
PaymentExpireMinutes int
ConfigSnapshot settingssecurity.OrderRiskControlConfig
}
// NormalizeRiskIP 生成游客风控键：IPv4 使用完整地址，IPv6 按 /64 前缀聚合。