feat(v2): limit_aware 策略(UsageSource 留位)+ Router 按配置派发 + config.Routing
This commit is contained in:
wangjia
@@ -18,6 +18,9 @@ type Config struct {
|
||||
Biz map[string]BizSystemConfig `mapstructure:"biz"`
|
||||
// Accounts 多账户配置注册表（v2）：凭证不写死配置，只存 env 前缀，真值运行时从环境变量取。
|
||||
Accounts []AccountConfig `mapstructure:"accounts"`
|
||||
// Routing 路由策略:channel → 策略名(round_robin/weighted/limit_aware)。
|
||||
// 缺省/未知一律当 round_robin(见 P5 计划 D3)。密钥无关,可写 config.yaml。
|
||||
Routing map[string]string `mapstructure:"routing"`
|
||||
}
|
||||
|
||||
// AccountConfig 单个收款账户的配置（v2 多账户路由用）。凭证严禁写进 config.yaml，
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
package accounts
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
)
|
||||
|
||||
// UsageSource 提供账户今日已收金额(minor),供 limit_aware 判定是否越 DailyLimit。
|
||||
// P5 内存实现(MemUsage)+ 生产空源(NopUsage);真实用量数据源由 P6 对账 job 提供,
|
||||
// 届时实现本接口注入 Router 即可,LimitAware/Router 不改(见计划 D5)。
|
||||
type UsageSource interface {
|
||||
TodayUsedMinor(accountID string) int64
|
||||
}
|
||||
|
||||
// NopUsage 恒返回 0:P6 前生产装配用,limit_aware 退化为基础策略(不误拒)。
|
||||
type NopUsage struct{}
|
||||
|
||||
func (NopUsage) TodayUsedMinor(string) int64 { return 0 }
|
||||
|
||||
// MemUsage 进程内用量计数(测试确定性 / 简单场景)。
|
||||
type MemUsage struct {
|
||||
mu sync.Mutex
|
||||
used map[string]int64
|
||||
}
|
||||
|
||||
func NewMemUsage() *MemUsage { return &MemUsage{used: map[string]int64{}} }
|
||||
|
||||
func (m *MemUsage) Add(accountID string, minor int64) {
|
||||
m.mu.Lock()
|
||||
m.used[accountID] += minor
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
func (m *MemUsage) TodayUsedMinor(accountID string) int64 {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.used[accountID]
|
||||
}
|
||||
|
||||
// LimitAware 过滤掉「DailyLimit>0 且 已用+本单 > DailyLimit」的账户,再把合格集交给
|
||||
// base 策略选(默认 round_robin)。全部越限 → ErrNoAccount(分摊避免单账户触风控冻结)。
|
||||
type LimitAware struct {
|
||||
base Strategy
|
||||
usage UsageSource
|
||||
}
|
||||
|
||||
func NewLimitAware(base Strategy, usage UsageSource) *LimitAware {
|
||||
if usage == nil {
|
||||
usage = NopUsage{}
|
||||
}
|
||||
return &LimitAware{base: base, usage: usage}
|
||||
}
|
||||
|
||||
func (l *LimitAware) Pick(key string, candidates []config.AccountConfig, hint PickHint) (config.AccountConfig, error) {
|
||||
eligible := make([]config.AccountConfig, 0, len(candidates))
|
||||
for _, c := range candidates {
|
||||
if c.DailyLimit > 0 && l.usage.TodayUsedMinor(c.AccountID)+hint.AmountMinor > c.DailyLimit {
|
||||
continue
|
||||
}
|
||||
eligible = append(eligible, c)
|
||||
}
|
||||
if len(eligible) == 0 {
|
||||
return config.AccountConfig{}, ErrNoAccount
|
||||
}
|
||||
return l.base.Pick(key, eligible, hint)
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
package accounts_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
)
|
||||
|
||||
func TestLimitAwareFiltersOverLimit(t *testing.T) {
|
||||
usage := accounts.NewMemUsage()
|
||||
usage.Add("a1", 9000) // a1 今日已用 9000
|
||||
la := accounts.NewLimitAware(accounts.NewRoundRobin(), usage)
|
||||
cs := []config.AccountConfig{
|
||||
{AccountID: "a1", DailyLimit: 10000}, // 9000+2000=11000 > 10000 → 排除
|
||||
{AccountID: "a2", DailyLimit: 10000}, // 0+2000 <= 10000 → 合格
|
||||
}
|
||||
for i := 0; i < 3; i++ {
|
||||
got, err := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 2000})
|
||||
if err != nil || got.AccountID != "a2" {
|
||||
t.Fatalf("越限的 a1 应被过滤、只剩 a2, got %v err %v", got.AccountID, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLimitAwareZeroLimitIsUnlimited(t *testing.T) {
|
||||
usage := accounts.NewMemUsage()
|
||||
usage.Add("a1", 1_000_000)
|
||||
la := accounts.NewLimitAware(accounts.NewRoundRobin(), usage)
|
||||
cs := []config.AccountConfig{{AccountID: "a1", DailyLimit: 0}} // 0=不限
|
||||
got, err := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 999})
|
||||
if err != nil || got.AccountID != "a1" {
|
||||
t.Fatalf("DailyLimit=0 应不限, got %v err %v", got.AccountID, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLimitAwareAllOverLimit(t *testing.T) {
|
||||
usage := accounts.NewMemUsage()
|
||||
usage.Add("a1", 10000)
|
||||
la := accounts.NewLimitAware(accounts.NewRoundRobin(), usage)
|
||||
cs := []config.AccountConfig{{AccountID: "a1", DailyLimit: 10000}}
|
||||
if _, err := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 1}); !errors.Is(err, accounts.ErrNoAccount) {
|
||||
t.Fatalf("全部越限应 ErrNoAccount, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNopUsageDegradesToBase(t *testing.T) {
|
||||
la := accounts.NewLimitAware(accounts.NewRoundRobin(), accounts.NopUsage{})
|
||||
cs := []config.AccountConfig{
|
||||
{AccountID: "a1", DailyLimit: 100}, {AccountID: "a2", DailyLimit: 100},
|
||||
}
|
||||
// 空用量源 → 谁都不越限 → 行为 = round_robin。
|
||||
// 注:AmountMinor 须 <= DailyLimit,否则即便 used=0 单笔也会越限(0+amount>limit),
|
||||
// 那样测的是「单笔超限」而非「NopUsage 退化」,与用例名/注释矛盾——故取 10 而非 999。
|
||||
g1, _ := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 10})
|
||||
g2, _ := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 10})
|
||||
if g1.AccountID != "a1" || g2.AccountID != "a2" {
|
||||
t.Fatalf("NopUsage 下应退化为轮询 a1,a2, got %s,%s", g1.AccountID, g2.AccountID)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
package accounts
|
||||
|
||||
import "github.com/wangjia/pay/config"
|
||||
|
||||
// Router 是 gateway 面向的 Picker 实现:取候选(EnabledFor)→ 按 hint 排除 → 派发策略。
|
||||
// 策略实例进程内单例(round_robin 计数器需长存);未配置/未知策略默认 round_robin。
|
||||
type Router struct {
|
||||
reg *Registry
|
||||
strategies map[string]string // channel -> 策略名
|
||||
byName map[string]Strategy
|
||||
def Strategy
|
||||
}
|
||||
|
||||
// NewRouter 装配路由。routing:channel→策略名(缺省/未知→round_robin);usage=nil→NopUsage。
|
||||
// limit_aware 复合 round_robin 作为合格集内的均摊基础策略。
|
||||
func NewRouter(reg *Registry, routing map[string]string, usage UsageSource) *Router {
|
||||
rr := NewRoundRobin()
|
||||
byName := map[string]Strategy{
|
||||
StrategyRoundRobin: rr,
|
||||
StrategyWeighted: NewWeighted(),
|
||||
StrategyLimitAware: NewLimitAware(rr, usage),
|
||||
}
|
||||
cp := map[string]string{}
|
||||
for k, v := range routing {
|
||||
cp[k] = v
|
||||
}
|
||||
return &Router{reg: reg, strategies: cp, byName: byName, def: rr}
|
||||
}
|
||||
|
||||
func (r *Router) strategyFor(channel string) Strategy {
|
||||
if name, ok := r.strategies[channel]; ok {
|
||||
if s, ok := r.byName[name]; ok {
|
||||
return s
|
||||
}
|
||||
}
|
||||
return r.def // 缺省 / 未知策略名 → round_robin(见计划 D3)
|
||||
}
|
||||
|
||||
// Pick 实现 accounts.Picker。
|
||||
func (r *Router) Pick(channel, region string, hint PickHint) (config.AccountConfig, error) {
|
||||
cands := r.reg.EnabledFor(channel, region)
|
||||
if len(cands) == 0 {
|
||||
return config.AccountConfig{}, ErrNoAccount
|
||||
}
|
||||
if len(hint.ExcludeAccounts) > 0 {
|
||||
if filtered := excludeAccounts(cands, hint.ExcludeAccounts); len(filtered) > 0 {
|
||||
cands = filtered // 过滤到空则保留全集:单账户渠道 retry 仍可用(见计划 D4)
|
||||
}
|
||||
}
|
||||
return r.strategyFor(channel).Pick(channel+"|"+region, cands, hint)
|
||||
}
|
||||
|
||||
func excludeAccounts(cands []config.AccountConfig, exclude []string) []config.AccountConfig {
|
||||
skip := make(map[string]struct{}, len(exclude))
|
||||
for _, id := range exclude {
|
||||
skip[id] = struct{}{}
|
||||
}
|
||||
out := make([]config.AccountConfig, 0, len(cands))
|
||||
for _, c := range cands {
|
||||
if _, ok := skip[c.AccountID]; ok {
|
||||
continue
|
||||
}
|
||||
out = append(out, c)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package accounts_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
)
|
||||
|
||||
func newReg() *accounts.Registry {
|
||||
return accounts.New([]config.AccountConfig{
|
||||
{AccountID: "f1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
|
||||
{AccountID: "f2", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
|
||||
{AccountID: "f3", Channel: "fake", Region: "global", Enabled: false, Weight: 1}, // 禁用不参与
|
||||
{AccountID: "a1", Channel: "alipay", Region: "cn", Enabled: true, Weight: 1},
|
||||
})
|
||||
}
|
||||
|
||||
func TestRouterDefaultRoundRobin(t *testing.T) {
|
||||
r := accounts.NewRouter(newReg(), nil, nil) // 无 routing → 默认 round_robin
|
||||
seq := []string{"f1", "f2", "f1"}
|
||||
for i, w := range seq {
|
||||
got, err := r.Pick("fake", "global", accounts.PickHint{OutTradeNo: "PAY-1"})
|
||||
if err != nil || got.AccountID != w {
|
||||
t.Fatalf("call %d got %v err %v want %s", i, got.AccountID, err, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRouterExcludeSwitchesAccount(t *testing.T) {
|
||||
r := accounts.NewRouter(newReg(), map[string]string{"fake": accounts.StrategyWeighted}, nil)
|
||||
base, _ := r.Pick("fake", "global", accounts.PickHint{OutTradeNo: "PAY-RETRY"})
|
||||
// retry:排除首选账户 → 必换到另一个。
|
||||
got, err := r.Pick("fake", "global", accounts.PickHint{OutTradeNo: "PAY-RETRY", ExcludeAccounts: []string{base.AccountID}})
|
||||
if err != nil || got.AccountID == base.AccountID {
|
||||
t.Fatalf("排除 %s 后应换账户, got %v err %v", base.AccountID, got.AccountID, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRouterExcludeEmptyFallsBackToAll(t *testing.T) {
|
||||
r := accounts.NewRouter(newReg(), nil, nil)
|
||||
// alipay/cn 只有 a1;排除 a1 后候选空 → 回退全集,仍返回 a1(单账户 retry 不失败)。
|
||||
got, err := r.Pick("alipay", "cn", accounts.PickHint{OutTradeNo: "PAY-2", ExcludeAccounts: []string{"a1"}})
|
||||
if err != nil || got.AccountID != "a1" {
|
||||
t.Fatalf("排除到空应回退全集, got %v err %v", got.AccountID, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRouterNoAccount(t *testing.T) {
|
||||
r := accounts.NewRouter(newReg(), nil, nil)
|
||||
if _, err := r.Pick("wechat", "cn", accounts.PickHint{}); !errors.Is(err, accounts.ErrNoAccount) {
|
||||
t.Fatalf("无账户渠道应 ErrNoAccount, got %v", err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user