docs(v2): P5 多账户路由策略计划(4 任务:Picker/round_robin/weighted/limit_aware骨架+gateway接线)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013nMthbVEmQquxBRKb9Fj8u
This commit is contained in:
wangjia
@@ -0,0 +1,929 @@
|
||||
# pay v2 · P5 多账户路由策略 Implementation Plan
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
> **设计文档(全景蓝图):** `docs/pay-v2-unified-gateway-design.html` §3.3(Account 多账户 + 路由:round_robin / weighted / limit_aware / by_region / crypto 地址池;**一单按策略选一个账户,Attempt 绑定 account_id,回调/退款/结算走同一账户**)。
|
||||
> **前置计划:** `docs/superpowers/plans/2026-07-10-pay-v2-p2-pipeline.md`(P2 已落地:`internal/provider`、`internal/gateway`、`internal/store` 扩展、`internal/accounts.Registry`)。本计划是 **P5**,替换 P2 里两处 `accts[0]`(取首个 enabled)的临时取舍。
|
||||
|
||||
**Goal:** 在 `internal/accounts` 上加一层**策略路由**:`Picker.Pick(channel, region, hint)` 按配置的 `route_strategy` 从「该渠道该区域的 enabled 账户」里选**一个**账户,取代 P2 的「首个 enabled」。落地三个策略 —— **round_robin**(进程内轮询计数器)、**weighted**(按 `Weight`,以 `hash(OutTradeNo)` 取模做**确定性**选择,不引不可控 rand)、**limit_aware**(按 `DailyLimit` 过滤超额账户 + 复合一个基础策略;用量数据源以 `UsageSource` 接口留位,本期内存实现 + 生产空源 = 退化为 round_robin,真实用量待 P6 对账)。`gateway.CreateOrder/RetryOrder` 改经 `Picker`;**retry 经 `hint.ExcludeAccounts` 换一个账户重试**(同渠道换号是本功能的真实价值:分摊限额 / 抗风控冻卡 / crypto 换地址)。
|
||||
|
||||
**Architecture:** 干净分层 —— `Registry.EnabledFor(channel, region)`(P1,负责过滤)只出**候选切片**;`Strategy.Pick(key, candidates, hint)` 是**纯函数式**选择器(round_robin 靠自身按 key 分桶的进程计数器、weighted 靠 `hash(OutTradeNo)`、limit_aware 靠 `UsageSource` + 组合基础策略),对候选切片操作 → **确定性、可单测、无 Registry 依赖**。`Router`(实现 `Picker`)负责:`EnabledFor` 取候选 → 按 `hint.ExcludeAccounts` 过滤(过滤到空则回退不过滤,保证单账户渠道 retry 不失败)→ 按 channel 的配置 `route_strategy` 派发到对应 `Strategy`。`gateway` 只持有 `accounts.Picker` 接口,不认具体策略。
|
||||
|
||||
**Tech Stack:** Go 1.26.1 · `github.com/wangjia/pay` · `internal/accounts`(P1 `Registry`/`config.AccountConfig`)· `internal/gateway`(P2)· `internal/store`(P1/P2 `OrderStore`)· `hash/fnv`(标准库,确定性 hash)· `sync`(进程计数器 mutex)。**全程 `:memory:`/纯内存单测,免 docker。**
|
||||
|
||||
## Global Constraints
|
||||
|
||||
- **复用 P1/P2,不重造**:候选过滤用 `accounts.Registry.EnabledFor(channel, region)`(已按 channel+enabled+region 过滤);账户模型 `config.AccountConfig`(`AccountID`/`Channel`/`Weight`/`Enabled`/`DailyLimit`/`Region`/`CredentialEnvPrefix`);gateway 现有签名/管线(`CreateOrder`/`RetryOrder`)只改「选账户」那一步,其余不动。
|
||||
- **确定性铁律**:测试禁 `rand`/`time.Now()` 做决策依赖。
|
||||
- **round_robin**:进程内计数器(`atomic`/mutex),**按调用顺序确定** → 测试串行调用断言 a1,a2,a1,a2。重启归零(进程级、不落库),**可接受**(见 D2)。
|
||||
- **weighted**:决策源 = `fnv32a(hint.OutTradeNo) % Σweight`,**同 OutTradeNo → 恒同账户**,跨重启/多节点一致,不注入 rand(见 D1 论证)。
|
||||
- **limit_aware**:用量来自注入的 `UsageSource`(内存 `MemUsage` 可 `Add` → 测试确定),不读时钟。
|
||||
- **Attempt 绑定选中账户**(设计 §3.3):`gateway` 建 Attempt 时 `AccountID = acct.AccountID`(P2 已如此),回调/退款/结算据此走同一账户 —— 本期不改这条,只改「acct 从哪来」。
|
||||
- **retry 换账户**:`RetryOrder` 收集该单该渠道**已试过的 account_id**,经 `hint.ExcludeAccounts` 传入 → Router 过滤后策略在**剩余**账户里选;剩余为空(单账户渠道)则回退用全集(retry 允许命中同账户,好过报错)。
|
||||
- **默认策略 = round_robin**(见 D3):配置缺省/空串一律当 round_robin;**单账户渠道下 round_robin ≡ weighted ≡ limit_aware ≡「首个 enabled」**(计数/权重/过滤对唯一候选都收敛到它),故默认切 round_robin 对现网单账户零行为变化、对多账户严格更优,无需配置迁移。
|
||||
- **crypto 地址池**(设计 §3.3):crypto 渠道下「account = 一个 xpub」,round_robin/weighted 选账户 = **地址来源轮换**;**地址派生在 P3 crypto adapter 的 `create()` 内**(消费 `provider.CreateRequest.Account`)。P5 只负责把 account 选对,不碰地址派生 —— 与 P3 接口(以 `CreateRequest.Account` 为界)不冲突。
|
||||
- 每步 `go build ./...` 通过;测试 `go test ./...`(纯内存,免 docker)。
|
||||
- 每任务严格 bite-sized TDD:写失败测试 → 跑失败 → 实现 → 跑通过 → commit。禁占位。
|
||||
|
||||
## 关键决策(执行前定案)
|
||||
|
||||
- **D1 weighted 确定性方案:hash(OutTradeNo) 取模,而非注入 rand。** 理由:①**可复现** —— 同一单无论在哪台节点、重启前后,选账户结果恒定,便于排障与对账("这单为什么落 a2"有确定答案);②**天然幂等友好** —— 无副作用、无共享随机态;③retry 换号由 `ExcludeAccounts` 显式驱动(见 D4),不依赖"下次随机不同",避免 rand 方案里"重试仍抽中同账户"的抖动。分布正确性靠大样本不同 OutTradeNo 的 hash 均匀性(fnv32a 足够),测试用 1000 个合成单号校验两账户都被选中 + 3:1 权重的粗略分布容差。
|
||||
- **D2 round_robin 状态:进程内计数器,重启归零、不落库、不跨节点共享。** 理由:轮询只为"大致均摊",不是强配额;落库/共享会给每次下单加一次写 + 锁争用,不值当。重启后从头轮询,长期仍均匀。**多实例部署下各实例独立轮询**,合起来仍近似均摊(可接受)。强配额语义由 limit_aware(D5)承担,不靠 round_robin。
|
||||
- **D3 默认策略 = round_robin(零值兼容)。** 见 Global Constraints;单账户渠道下与 P2「首个 enabled」逐位等价,现网零行为变化。
|
||||
- **D4 retry 换账户 = `hint.ExcludeAccounts`。** `RetryOrder` 传入本单本渠道已试账户集,Router 在派发策略前先从候选里剔除它们 → **任何策略**都自动在"没试过的账户"里选,换号是路由层通用能力,不需每个策略各自实现。过滤到空 → 回退全集(单账户渠道 retry 仍可用)。
|
||||
- **D5 limit_aware 取舍:本期实现内存计数骨架 + `UsageSource` 接口留 DB 位(方案 a),不延到 P6。** 理由:①**逻辑本期就该被测到** —— "used+amount 是否越 DailyLimit → 过滤该账户 → 在合格集里走基础策略"这套判定,靠注入 `MemUsage` 即可确定性单测,不必等真实数据;②**安全退化** —— 生产装配传空源(`NopUsage`,恒返回 0 已用),则所有账户永不越限 → limit_aware **行为等同 round_robin**,不会因为"P6 未就绪"而错误拒单;③**P6 零改接口接入** —— 对账 job 就绪后,实现一个读对账用量的 `UsageSource` 注入 Router 即可,`LimitAware`/`Router` 一行不改。相比"纯枚举保留、逻辑延到 P6"(方案 b),方案 a 让该策略**现在就可用且被测**,且 P6 只补数据源不补逻辑,风险更低。**明确不做**:真实用量数据源(P6 对账提供)、跨 region 故障转移(later)。
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Picker/Strategy 核心类型 + round_robin 策略
|
||||
|
||||
**Files:**
|
||||
- Create: `internal/accounts/picker.go`(`Picker`/`Strategy` 接口 + `PickHint` + 策略名常量 + `ErrNoAccount` + `RoundRobin`)
|
||||
- Test: `internal/accounts/picker_test.go`
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: `config.AccountConfig`(P1)。
|
||||
- Produces:
|
||||
- `var ErrNoAccount = errors.New("accounts: no candidate account for channel/region")`
|
||||
- `type PickHint struct{ OutTradeNo string; AmountMinor int64; ExcludeAccounts []string }`(OutTradeNo=weighted 确定性锚点 + 地址池亲和;AmountMinor=limit_aware 判定;ExcludeAccounts=retry 换号)。
|
||||
- `type Picker interface{ Pick(channel, region string, hint PickHint) (config.AccountConfig, error) }`(gateway 面向此接口)。
|
||||
- `type Strategy interface{ Pick(key string, candidates []config.AccountConfig, hint PickHint) (config.AccountConfig, error) }`(**纯函数式**:对已过滤候选选一个;`key = channel+"|"+region`,round_robin 用它分桶计数,其余策略忽略)。
|
||||
- 策略名常量:`StrategyRoundRobin = "round_robin"`、`StrategyWeighted = "weighted"`、`StrategyLimitAware = "limit_aware"`。
|
||||
- `type RoundRobin struct{...}` · `func NewRoundRobin() *RoundRobin`(实现 `Strategy`,进程内按 key 分桶轮询计数器,mutex 保护)。
|
||||
|
||||
- [ ] **Step 1: 写失败测试**
|
||||
|
||||
`internal/accounts/picker_test.go`:
|
||||
```go
|
||||
package accounts_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
)
|
||||
|
||||
func cands(ids ...string) []config.AccountConfig {
|
||||
out := make([]config.AccountConfig, 0, len(ids))
|
||||
for _, id := range ids {
|
||||
out = append(out, config.AccountConfig{AccountID: id, Channel: "fake", Region: "global", Enabled: true, Weight: 1})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func TestRoundRobinCyclesInOrder(t *testing.T) {
|
||||
rr := accounts.NewRoundRobin()
|
||||
cs := cands("a1", "a2", "a3")
|
||||
want := []string{"a1", "a2", "a3", "a1", "a2"}
|
||||
for i, w := range want {
|
||||
got, err := rr.Pick("fake|global", cs, accounts.PickHint{})
|
||||
if err != nil || got.AccountID != w {
|
||||
t.Fatalf("call %d: got %v err %v, want %s", i, got.AccountID, err, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoundRobinKeysAreIndependent(t *testing.T) {
|
||||
rr := accounts.NewRoundRobin()
|
||||
cs := cands("a1", "a2")
|
||||
// 不同 key(渠道/区域)各自独立计数,互不串扰。
|
||||
g1, _ := rr.Pick("alipay|cn", cs, accounts.PickHint{})
|
||||
g2, _ := rr.Pick("wechat|cn", cs, accounts.PickHint{})
|
||||
if g1.AccountID != "a1" || g2.AccountID != "a1" {
|
||||
t.Fatalf("每个 key 首次都应从 a1 起, got %s / %s", g1.AccountID, g2.AccountID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStrategyEmptyCandidates(t *testing.T) {
|
||||
rr := accounts.NewRoundRobin()
|
||||
if _, err := rr.Pick("fake|global", nil, accounts.PickHint{}); !errors.Is(err, accounts.ErrNoAccount) {
|
||||
t.Fatalf("空候选应 ErrNoAccount, got %v", err)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go test ./internal/accounts/ -run 'RoundRobin|StrategyEmpty' -v`
|
||||
Expected: 编译失败 —— `Picker`/`Strategy`/`RoundRobin` 未定义。
|
||||
|
||||
- [ ] **Step 3: 写实现**
|
||||
|
||||
`internal/accounts/picker.go`:
|
||||
```go
|
||||
package accounts
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
)
|
||||
|
||||
// ErrNoAccount 表示该 channel/region 下没有可选账户(候选为空,或被策略全部排除)。
|
||||
var ErrNoAccount = errors.New("accounts: no candidate account for channel/region")
|
||||
|
||||
// 路由策略名(配置 routing.<channel> 的取值;缺省/未知一律当 round_robin)。
|
||||
const (
|
||||
StrategyRoundRobin = "round_robin"
|
||||
StrategyWeighted = "weighted"
|
||||
StrategyLimitAware = "limit_aware"
|
||||
)
|
||||
|
||||
// PickHint 选账户的上下文提示(确定性 + retry 换号)。
|
||||
type PickHint struct {
|
||||
OutTradeNo string // weighted 的确定性锚点;crypto 地址池亲和
|
||||
AmountMinor int64 // limit_aware 判定 used+amount<=DailyLimit
|
||||
ExcludeAccounts []string // retry:排除已试过的账户,换一个重试
|
||||
}
|
||||
|
||||
// Picker 面向 gateway 的路由入口:按渠道配置的策略从 enabled 账户里选一个。
|
||||
type Picker interface {
|
||||
Pick(channel, region string, hint PickHint) (config.AccountConfig, error)
|
||||
}
|
||||
|
||||
// Strategy 是纯函数式选择器:对「已过滤的候选切片」选一个账户。
|
||||
// key = channel+"|"+region,round_robin 用它分桶进程计数器,其余策略忽略。
|
||||
type Strategy interface {
|
||||
Pick(key string, candidates []config.AccountConfig, hint PickHint) (config.AccountConfig, error)
|
||||
}
|
||||
|
||||
// RoundRobin 进程内轮询:按 key 分桶各自递增计数器,mod 候选数。
|
||||
// 重启归零、不落库、不跨实例共享(见计划 D2);只为均摊,不作强配额。
|
||||
type RoundRobin struct {
|
||||
mu sync.Mutex
|
||||
counters map[string]uint64
|
||||
}
|
||||
|
||||
func NewRoundRobin() *RoundRobin { return &RoundRobin{counters: map[string]uint64{}} }
|
||||
|
||||
func (r *RoundRobin) Pick(key string, candidates []config.AccountConfig, _ PickHint) (config.AccountConfig, error) {
|
||||
if len(candidates) == 0 {
|
||||
return config.AccountConfig{}, ErrNoAccount
|
||||
}
|
||||
r.mu.Lock()
|
||||
n := r.counters[key]
|
||||
r.counters[key] = n + 1
|
||||
r.mu.Unlock()
|
||||
return candidates[int(n%uint64(len(candidates)))], nil
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 跑测试确认通过**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go test ./internal/accounts/ -v`
|
||||
Expected: 新增 3 测试 + P1 既有 `TestEnabledForAndCredential` 全 PASS。
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
cd /Users/wangjia/code/pay
|
||||
git add internal/accounts/picker.go internal/accounts/picker_test.go
|
||||
git commit -m "feat(v2): 账户路由 Picker/Strategy 接口 + round_robin 策略(替换首个 enabled)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: weighted 策略(按 Weight,hash(OutTradeNo) 确定性)
|
||||
|
||||
**Files:**
|
||||
- Create: `internal/accounts/weighted.go`(`Weighted` 策略)
|
||||
- Test: `internal/accounts/weighted_test.go`
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: `config.AccountConfig`、`PickHint`、`Strategy`(Task 1)。
|
||||
- Produces:
|
||||
- `type Weighted struct{}` · `func NewWeighted() *Weighted`(实现 `Strategy`)。
|
||||
- 决策:`target = fnv32a(hint.OutTradeNo) % Σweight`,按候选顺序累加 `weight` 命中区间;`Weight<=0` 记 1(对齐 gorm `default:1`)。`OutTradeNo` 空 → 退化为取首个(确定,不 panic)。
|
||||
|
||||
- [ ] **Step 1: 写失败测试**
|
||||
|
||||
`internal/accounts/weighted_test.go`:
|
||||
```go
|
||||
package accounts_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
"github.com/wangjia/pay/internal/accounts"
|
||||
)
|
||||
|
||||
func TestWeightedDeterministic(t *testing.T) {
|
||||
w := accounts.NewWeighted()
|
||||
cs := cands("a1", "a2", "a3")
|
||||
// 同一 OutTradeNo 必落同一账户(确定性,可复现)。
|
||||
first, _ := w.Pick("fake|global", cs, accounts.PickHint{OutTradeNo: "PAY-XYZ"})
|
||||
for i := 0; i < 5; i++ {
|
||||
got, _ := w.Pick("fake|global", cs, accounts.PickHint{OutTradeNo: "PAY-XYZ"})
|
||||
if got.AccountID != first.AccountID {
|
||||
t.Fatalf("同单号应恒定命中 %s, 第 %d 次得 %s", first.AccountID, i, got.AccountID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeightedDistributesAndRespectsWeight(t *testing.T) {
|
||||
w := accounts.NewWeighted()
|
||||
// a1 权重 3,a2 权重 1 → 长跑大致 3:1。
|
||||
cs := []config.AccountConfig{
|
||||
{AccountID: "a1", Weight: 3},
|
||||
{AccountID: "a2", Weight: 1},
|
||||
}
|
||||
counts := map[string]int{}
|
||||
for i := 0; i < 4000; i++ {
|
||||
no := "PAY-" + string(rune('A'+i%26)) + string(rune('0'+i/26%10)) + string(rune('a'+i/260%26)) + string(rune('0'+i%7))
|
||||
got, _ := w.Pick("fake|global", cs, accounts.PickHint{OutTradeNo: no})
|
||||
counts[got.AccountID]++
|
||||
}
|
||||
if counts["a1"] == 0 || counts["a2"] == 0 {
|
||||
t.Fatalf("两账户都应被选中, got %+v", counts)
|
||||
}
|
||||
// a1 应显著多于 a2(3:1 容差:a1 至少是 a2 的 2 倍)。
|
||||
if counts["a1"] < counts["a2"]*2 {
|
||||
t.Fatalf("权重 3:1 未体现, got %+v", counts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWeightedZeroWeightTreatedAsOne(t *testing.T) {
|
||||
w := accounts.NewWeighted()
|
||||
cs := []config.AccountConfig{{AccountID: "a1", Weight: 0}, {AccountID: "a2", Weight: 0}}
|
||||
counts := map[string]int{}
|
||||
for i := 0; i < 400; i++ {
|
||||
got, _ := w.Pick("fake|global", cs, accounts.PickHint{OutTradeNo: "N" + string(rune('0'+i%10)) + string(rune('a'+i%26))})
|
||||
counts[got.AccountID]++
|
||||
}
|
||||
if counts["a1"] == 0 || counts["a2"] == 0 {
|
||||
t.Fatalf("weight=0 应当 1 处理、两账户都可选, got %+v", counts)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go test ./internal/accounts/ -run Weighted -v`
|
||||
Expected: 编译失败 —— `Weighted` 未定义。
|
||||
|
||||
- [ ] **Step 3: 写实现**
|
||||
|
||||
`internal/accounts/weighted.go`:
|
||||
```go
|
||||
package accounts
|
||||
|
||||
import (
|
||||
"hash/fnv"
|
||||
|
||||
"github.com/wangjia/pay/config"
|
||||
)
|
||||
|
||||
// Weighted 按 Weight 加权选账户,决策源为 hash(OutTradeNo)(确定性、可复现、跨节点一致,
|
||||
// 见计划 D1)。Weight<=0 记 1(对齐 model.Account gorm default:1)。retry 换号由
|
||||
// Router 的 ExcludeAccounts 过滤驱动,不依赖随机。
|
||||
type Weighted struct{}
|
||||
|
||||
func NewWeighted() *Weighted { return &Weighted{} }
|
||||
|
||||
func (Weighted) Pick(_ string, candidates []config.AccountConfig, hint PickHint) (config.AccountConfig, error) {
|
||||
if len(candidates) == 0 {
|
||||
return config.AccountConfig{}, ErrNoAccount
|
||||
}
|
||||
var total uint32
|
||||
for _, c := range candidates {
|
||||
total += weightOf(c)
|
||||
}
|
||||
if total == 0 { // 理论不达(weightOf>=1),兜底
|
||||
return candidates[0], nil
|
||||
}
|
||||
h := fnv.New32a()
|
||||
_, _ = h.Write([]byte(hint.OutTradeNo))
|
||||
target := h.Sum32() % total
|
||||
var acc uint32
|
||||
for _, c := range candidates {
|
||||
acc += weightOf(c)
|
||||
if target < acc {
|
||||
return c, nil
|
||||
}
|
||||
}
|
||||
return candidates[len(candidates)-1], nil // 浮点无关的整数走位,兜底不达
|
||||
}
|
||||
|
||||
func weightOf(c config.AccountConfig) uint32 {
|
||||
if c.Weight <= 0 {
|
||||
return 1
|
||||
}
|
||||
return uint32(c.Weight)
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 跑测试确认通过**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go test ./internal/accounts/ -v`
|
||||
Expected: weighted 3 测试 + Task 1 + P1 全 PASS。
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
cd /Users/wangjia/code/pay
|
||||
git add internal/accounts/weighted.go internal/accounts/weighted_test.go
|
||||
git commit -m "feat(v2): weighted 路由策略(按 Weight,hash(OutTradeNo) 确定性)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: limit_aware 策略 + UsageSource + Router(按配置派发)
|
||||
|
||||
**Files:**
|
||||
- Create: `internal/accounts/limit_aware.go`(`UsageSource`/`NopUsage`/`MemUsage` + `LimitAware`)
|
||||
- Create: `internal/accounts/router.go`(`Router` 实现 `Picker`:候选取用 + Exclude 过滤 + 策略派发)
|
||||
- Edit: `config/config.go`(`Config` 增 `Routing map[string]string`)
|
||||
- Test: `internal/accounts/limit_aware_test.go`
|
||||
- Test: `internal/accounts/router_test.go`
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: `Registry.EnabledFor`(P1)、`Strategy`/`RoundRobin`/`Weighted`(Task 1/2)、`config.AccountConfig`。
|
||||
- Produces:
|
||||
- `type UsageSource interface{ TodayUsedMinor(accountID string) int64 }`(今日已收金额,minor)。
|
||||
- `type NopUsage struct{}` → 恒 0(生产装配用,P6 前 limit_aware 退化为基础策略)。
|
||||
- `type MemUsage struct{...}` · `func NewMemUsage() *MemUsage` · `Add(accountID string, minor int64)`(测试/内存计数)。
|
||||
- `type LimitAware struct{ base Strategy; usage UsageSource }` · `func NewLimitAware(base Strategy, usage UsageSource) *LimitAware`(过滤 `DailyLimit>0 && used+amount>limit` 的账户,再交 `base` 选;全越限 → `ErrNoAccount`)。
|
||||
- `type Router struct{...}` · `func NewRouter(reg *Registry, routing map[string]string, usage UsageSource) *Router`(实现 `Picker`)。routing:channel→策略名;缺省/未知→round_robin。usage=nil→`NopUsage`。
|
||||
- `config.Config` 增 `Routing map[string]string mapstructure:"routing"`。
|
||||
|
||||
- [ ] **Step 1: 写失败测试**
|
||||
|
||||
`internal/accounts/limit_aware_test.go`:
|
||||
```go
|
||||
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。
|
||||
g1, _ := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 999})
|
||||
g2, _ := la.Pick("fake|global", cs, accounts.PickHint{AmountMinor: 999})
|
||||
if g1.AccountID != "a1" || g2.AccountID != "a2" {
|
||||
t.Fatalf("NopUsage 下应退化为轮询 a1,a2, got %s,%s", g1.AccountID, g2.AccountID)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
`internal/accounts/router_test.go`:
|
||||
```go
|
||||
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)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go test ./internal/accounts/ -run 'LimitAware|NopUsage|Router' -v`
|
||||
Expected: 编译失败 —— `LimitAware`/`MemUsage`/`Router` 未定义。
|
||||
|
||||
- [ ] **Step 3: 写 limit_aware 实现**
|
||||
|
||||
`internal/accounts/limit_aware.go`:
|
||||
```go
|
||||
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)
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 写 Router 实现**
|
||||
|
||||
`internal/accounts/router.go`:
|
||||
```go
|
||||
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
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 5: config 增 Routing 字段**
|
||||
|
||||
`config/config.go` 的 `Config` struct 末尾追加(在 `Accounts []AccountConfig` 之后):
|
||||
```go
|
||||
// Routing 路由策略:channel → 策略名(round_robin/weighted/limit_aware)。
|
||||
// 缺省/未知一律当 round_robin(见 P5 计划 D3)。密钥无关,可写 config.yaml。
|
||||
Routing map[string]string `mapstructure:"routing"`
|
||||
```
|
||||
|
||||
- [ ] **Step 6: 跑测试确认通过**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go build ./... && go test ./internal/accounts/ -v`
|
||||
Expected: limit_aware 4 + router 4 + Task1/2 + P1 全 PASS;`config` 编译通过。
|
||||
|
||||
- [ ] **Step 7: Commit**
|
||||
|
||||
```bash
|
||||
cd /Users/wangjia/code/pay
|
||||
git add internal/accounts/limit_aware.go internal/accounts/router.go internal/accounts/limit_aware_test.go internal/accounts/router_test.go config/config.go
|
||||
git commit -m "feat(v2): limit_aware 策略(UsageSource 留位)+ Router 按配置派发 + config.Routing"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 4: gateway 接入 Picker(CreateOrder/RetryOrder 换号)+ store 查已试账户 + 装配
|
||||
|
||||
**Files:**
|
||||
- Edit: `internal/gateway/gateway.go`(`Gateway.accounts *accounts.Registry` → `picker accounts.Picker`;两处 `accts[0]` → `picker.Pick`;retry 传 `ExcludeAccounts`)
|
||||
- Create: `internal/store/attempt_accounts.go`(`AttemptAccountIDs(outTradeNo, channel)`)
|
||||
- Edit: `main.go`(装配 `accounts.NewRouter` 传入 gateway)
|
||||
- Edit: `internal/gateway/gateway_test.go`(`newGateway` 用 Router 构造)
|
||||
- Test: `internal/store/attempt_accounts_test.go`
|
||||
- Test: `internal/gateway/gateway_test.go` 增 retry 换号断言
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: `accounts.Picker`/`NewRouter`/`PickHint`(Task 1-3)、`store.OrderStore`、`config.C.Accounts`/`config.C.Routing`。
|
||||
- Produces:
|
||||
- `store.OrderStore.AttemptAccountIDs(outTradeNo, channel string) ([]string, error)` —— 该单该渠道所有 attempt 的 distinct account_id(供 retry 排除)。
|
||||
- `gateway.New(orders, providers, picker accounts.Picker, products, webhook, region)` —— 第 3 参由 `*accounts.Registry` 改 `accounts.Picker`。
|
||||
- `CreateOrder`/`RetryOrder` 经 `picker.Pick`;`ErrNoAccount` 映射保持(`errors.Is(err, accounts.ErrNoAccount)` → `gateway.ErrNoAccount`)。
|
||||
|
||||
- [ ] **Step 1: 写失败测试(store)**
|
||||
|
||||
`internal/store/attempt_accounts_test.go`:
|
||||
```go
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
func TestAttemptAccountIDs(t *testing.T) {
|
||||
s := store.NewOrderStore(model.OpenTestDB(t))
|
||||
if err := s.CreateOrder(&model.OrderV2{OutTradeNo: "PAY-AC1", Subject: "x", AmountMinor: 100, Currency: "USDT", Status: model.OrderPendingV2}); err != nil {
|
||||
t.Fatalf("order: %v", err)
|
||||
}
|
||||
for _, tc := range []struct{ ref, acct string }{{"R1", "a1"}, {"R2", "a2"}, {"R3", "a1"}} {
|
||||
if err := s.CreateAttempt(&model.Attempt{
|
||||
OutTradeNo: "PAY-AC1", Channel: "fake", AccountID: tc.acct, Provider: "fake",
|
||||
ProviderRef: tc.ref, RenderType: "crypto_address", AmountMinor: 100, Currency: "USDT",
|
||||
Status: model.AttemptPending,
|
||||
}); err != nil {
|
||||
t.Fatalf("attempt %s: %v", tc.ref, err)
|
||||
}
|
||||
}
|
||||
ids, err := s.AttemptAccountIDs("PAY-AC1", "fake")
|
||||
if err != nil {
|
||||
t.Fatalf("AttemptAccountIDs: %v", err)
|
||||
}
|
||||
set := map[string]bool{}
|
||||
for _, id := range ids {
|
||||
set[id] = true
|
||||
}
|
||||
if len(set) != 2 || !set["a1"] || !set["a2"] {
|
||||
t.Fatalf("应含 distinct a1,a2, got %v", ids)
|
||||
}
|
||||
// 其它渠道 / 其它单不串。
|
||||
if got, _ := s.AttemptAccountIDs("PAY-AC1", "alipay"); len(got) != 0 {
|
||||
t.Fatalf("别的渠道应空, got %v", got)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go test ./internal/store/ -run AttemptAccountIDs -v`
|
||||
Expected: 编译失败 —— `AttemptAccountIDs` 未定义。
|
||||
|
||||
- [ ] **Step 3: 写 store 实现**
|
||||
|
||||
`internal/store/attempt_accounts.go`:
|
||||
```go
|
||||
package store
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
)
|
||||
|
||||
// AttemptAccountIDs returns the distinct account_ids already tried for an order on a
|
||||
// channel (used by retry to switch to a not-yet-tried account via PickHint.ExcludeAccounts).
|
||||
func (s *OrderStore) AttemptAccountIDs(outTradeNo, channel string) ([]string, error) {
|
||||
var ids []string
|
||||
if err := s.db.Model(&model.Attempt{}).
|
||||
Where("out_trade_no = ? AND channel = ? AND account_id <> ''", outTradeNo, channel).
|
||||
Distinct().Pluck("account_id", &ids).Error; err != nil {
|
||||
return nil, fmt.Errorf("store.AttemptAccountIDs: %w", err)
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 跑测试确认通过(store)**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go test ./internal/store/ -v`
|
||||
Expected: 新增 + P1/P2 既有全 PASS。
|
||||
|
||||
- [ ] **Step 5: 改 gateway 接入 Picker**
|
||||
|
||||
`internal/gateway/gateway.go`:①`import` 保留(已含 `errors`、`accounts`);②`Gateway` 字段 `accounts *accounts.Registry` 改为 `picker accounts.Picker`;③`New` 第 3 参改类型;④两处选账户改经 picker。
|
||||
|
||||
`Gateway` struct + `New`:
|
||||
```go
|
||||
type Gateway struct {
|
||||
orders *store.OrderStore
|
||||
providers *provider.Registry
|
||||
picker accounts.Picker
|
||||
products ProductResolver
|
||||
webhook WebhookEnqueuer
|
||||
region string
|
||||
}
|
||||
|
||||
func New(orders *store.OrderStore, providers *provider.Registry, picker accounts.Picker,
|
||||
products ProductResolver, webhook WebhookEnqueuer, region string) *Gateway {
|
||||
return &Gateway{orders: orders, providers: providers, picker: picker,
|
||||
products: products, webhook: webhook, region: region}
|
||||
}
|
||||
```
|
||||
|
||||
`CreateOrder` —— 把
|
||||
```go
|
||||
accts := g.accounts.EnabledFor(in.Method, g.region)
|
||||
if len(accts) == 0 {
|
||||
return nil, ErrNoAccount
|
||||
}
|
||||
acct := accts[0] // 路由策略(round_robin/weighted/…)在 P5;P2 取首个 enabled。
|
||||
|
||||
outNo := util.NewOutTradeNo("pay")
|
||||
```
|
||||
改为(先生成单号,再据单号确定性选账户):
|
||||
```go
|
||||
outNo := util.NewOutTradeNo("pay")
|
||||
acct, err := g.picker.Pick(in.Method, g.region, accounts.PickHint{
|
||||
OutTradeNo: outNo, AmountMinor: amountMinor,
|
||||
})
|
||||
if err != nil {
|
||||
if errors.Is(err, accounts.ErrNoAccount) {
|
||||
return nil, ErrNoAccount
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
```
|
||||
(注意:`outNo` 声明整体上移到 `acct` 之前;下方 `CreateOrder(&model.OrderV2{OutTradeNo: outNo...})` 不变。)
|
||||
|
||||
`RetryOrder` —— 把
|
||||
```go
|
||||
accts := g.accounts.EnabledFor(method, g.region)
|
||||
if len(accts) == 0 {
|
||||
return nil, ErrNoAccount
|
||||
}
|
||||
acct := accts[0]
|
||||
|
||||
if _, err := g.orders.ExpirePendingAttempts(outTradeNo); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
```
|
||||
改为(收集已试账户 → 换号选;先算排除集再废旧尝试,二者互不依赖):
|
||||
```go
|
||||
tried, err := g.orders.AttemptAccountIDs(outTradeNo, method)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
acct, err := g.picker.Pick(method, g.region, accounts.PickHint{
|
||||
OutTradeNo: outTradeNo, AmountMinor: o.AmountMinor, ExcludeAccounts: tried,
|
||||
})
|
||||
if err != nil {
|
||||
if errors.Is(err, accounts.ErrNoAccount) {
|
||||
return nil, ErrNoAccount
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if _, err := g.orders.ExpirePendingAttempts(outTradeNo); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
```
|
||||
> `RetryOrder` 里 `prov, err := g.providers.Get(method)` 之前已用 `err`;上面新增块复用同名 `err`(用 `=` 而非 `:=` 对已声明变量;`tried, err :=` 首次声明 `tried` 用 `:=` 合法)。实现时确认 `err` 作用域,如与既有 `err` 冲突按 Go 规则调整为 `:=`/`=`。
|
||||
|
||||
- [ ] **Step 6: 改 gateway_test 装配 + 增 retry 换号断言**
|
||||
|
||||
`internal/gateway/gateway_test.go` 的 `newGateway`:把
|
||||
```go
|
||||
areg := accounts.New([]config.AccountConfig{
|
||||
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
|
||||
})
|
||||
spy := &spyEnqueuer{}
|
||||
g := gateway.New(orders, preg, areg, stubResolver{}, spy, "global")
|
||||
```
|
||||
改为(**两个账户**,好测 retry 换号 + round_robin;Router 默认 round_robin):
|
||||
```go
|
||||
areg := accounts.New([]config.AccountConfig{
|
||||
{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
|
||||
{AccountID: "fake-a2", Channel: "fake", Region: "global", Enabled: true, Weight: 1},
|
||||
})
|
||||
picker := accounts.NewRouter(areg, nil, nil) // 默认 round_robin
|
||||
spy := &spyEnqueuer{}
|
||||
g := gateway.New(orders, preg, picker, stubResolver{}, spy, "global")
|
||||
```
|
||||
> 既有 `TestCreateOrderPipeline` 断言 attempt `AccountID == "fake-a1"`:round_robin 首个 key 首次仍返回 `fake-a1`(候选顺序 a1,a2,计数 0),断言不变即通过。既有 `TestRetryAndCancel` 断言 retry 后恰 1 个 pending 尝试,仍成立。
|
||||
|
||||
新增测试(retry 必换到未试过的账户):
|
||||
```go
|
||||
func TestRetrySwitchesAccount(t *testing.T) {
|
||||
g, _, _, orders := newGateway(t)
|
||||
ctx := context.Background()
|
||||
res, _ := g.CreateOrder(ctx, gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
|
||||
|
||||
// 首单落 fake-a1(round_robin 计数 0);retry 排除 a1 → 必落 fake-a2。
|
||||
if _, err := g.RetryOrder(ctx, res.OrderNo, "fake"); err != nil {
|
||||
t.Fatalf("retry: %v", err)
|
||||
}
|
||||
pend, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
|
||||
if len(pend) != 1 || pend[0].AccountID != "fake-a2" {
|
||||
t.Fatalf("retry 应换到 fake-a2, got %+v", pend)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- [ ] **Step 7: 改 main.go 装配 Router**
|
||||
|
||||
`main.go` 把
|
||||
```go
|
||||
acctReg := accounts.New(config.C.Accounts)
|
||||
gw := gateway.New(orderStore, pReg, acctReg, productResolver, notifier, "cn")
|
||||
```
|
||||
改为:
|
||||
```go
|
||||
acctReg := accounts.New(config.C.Accounts)
|
||||
// P5 多账户路由:按 config.routing.<channel> 选策略(缺省 round_robin)。
|
||||
// limit_aware 用量数据源 P6 对账就绪前用空源(NopUsage,退化为 round_robin)。
|
||||
acctPicker := accounts.NewRouter(acctReg, config.C.Routing, accounts.NopUsage{})
|
||||
gw := gateway.New(orderStore, pReg, acctPicker, productResolver, notifier, "cn")
|
||||
```
|
||||
> `acctReg` 仍需构造并传入 `NewRouter`(Registry 提供 `EnabledFor` + P3 的 `Credential`);其它渠道 adapter 装配若也用 `acctReg` 不受影响。
|
||||
|
||||
- [ ] **Step 8: 全量编译 + 测试**
|
||||
|
||||
Run: `cd /Users/wangjia/code/pay && go build ./... && go test ./...`
|
||||
Expected: 全部包 `ok`(accounts/gateway/store/handler/provider/... 全绿)。
|
||||
|
||||
- [ ] **Step 9: Commit**
|
||||
|
||||
```bash
|
||||
cd /Users/wangjia/code/pay
|
||||
git add internal/gateway/gateway.go internal/gateway/gateway_test.go internal/store/attempt_accounts.go internal/store/attempt_accounts_test.go main.go
|
||||
git commit -m "feat(v2): gateway 经 Picker 路由选账户 + retry 换号(ExcludeAccounts);替换首个 enabled"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Self-Review
|
||||
|
||||
**Spec coverage(P5 范围,对照设计 §3.3):**
|
||||
- 三层「channel → account(N) → 每单按策略选一个」:`Registry.EnabledFor` 出候选 → `Router` 派发 `Strategy` 选一个(Task 3)= 设计 §3.3 ✓。
|
||||
- round_robin(轮询分摊,Task 1)/ weighted(按 Weight,Task 2)/ limit_aware(按 DailyLimit 分摊避免超额触风控,Task 3)三策略落地 ✓;`by_region / amount / biz` 未做 —— region 已由 `EnabledFor` 承担粗过滤,amount/biz 维度**明确延后**(设计列作可选,非本期硬需求)。
|
||||
- crypto 地址池:crypto 渠道 round_robin/weighted 选账户 = 地址来源轮换;**地址派生在 P3 crypto adapter `create()`**(消费 `CreateRequest.Account`),P5 只选账户,与 P3(以 `CreateRequest.Account` 为界)不冲突(Global Constraints)✓。
|
||||
- Attempt 绑定选中 account_id(P2 已建、本期沿用),回调/退款/结算走同一账户 ✓。
|
||||
- 替换 P2 两处 `accts[0]`(CreateOrder/RetryOrder)为 `picker.Pick`(Task 4)✓。
|
||||
|
||||
**复用 P1/P2(不重造):** `accounts.Registry.EnabledFor`(候选过滤)、`config.AccountConfig`(全字段 Weight/DailyLimit/Region)、`gateway` 管线与 `store.OrderStore` 既有方法、`model.Attempt.AccountID` 绑定 —— 全直接复用。新增仅:`Picker`/`Strategy`/三策略/`Router`/`UsageSource`(accounts)、`store.AttemptAccountIDs`、`config.Routing` 字段、gateway 选账户一步的改线。
|
||||
|
||||
**确定性(禁 rand/time.Now 决策):** round_robin=进程计数器(调用顺序确定)· weighted=`fnv32a(OutTradeNo)`(同单恒同账户)· limit_aware=注入 `MemUsage`(测试可控)· retry 换号=显式 `ExcludeAccounts`(不靠"下次随机不同")。全部单测纯内存、免 docker。
|
||||
|
||||
**关键决策落定:** D1 weighted 用 hash(OutTradeNo) 而非注入 rand(可复现/幂等友好)· D2 round_robin 进程内计数器重启归零(只均摊不强配额,可接受)· D3 默认 round_robin(单账户渠道与 P2「首个 enabled」逐位等价,现网零变化)· D4 retry 换号走路由层 `ExcludeAccounts`(通用于所有策略,过滤到空回退全集)· D5 limit_aware 本期实现内存计数骨架 + `UsageSource` 接口留 DB 位、生产空源退化为 round_robin、P6 零改接口接真实用量。
|
||||
|
||||
**Type consistency:** `accounts.Picker`(Router 实现)被 `gateway.Gateway` 持有;`accounts.Strategy`(RoundRobin/Weighted/LimitAware 实现)被 Router 派发;`PickHint{OutTradeNo,AmountMinor,ExcludeAccounts}` 在 Task 1 定义、Task 2-4 一致消费;`ErrNoAccount` 由 accounts 抛出、gateway 经 `errors.Is` 映射为 `gateway.ErrNoAccount`(handler 的 HTTP `no_account` 映射不变)。金额一律 `int64 minor`。
|
||||
|
||||
**已知 scope 取舍(记录,不阻塞 P5):**
|
||||
- limit_aware 真实用量数据源在 P6(对账);本期 `NopUsage` 生产装配 → 退化 round_robin,不误拒。
|
||||
- round_robin 计数器不跨实例共享(多副本部署各自轮询,合起来近似均摊);强配额靠 limit_aware。
|
||||
- `by_region / by_amount / by_biz` 策略未做(region 已由 `EnabledFor` 粗过滤;amount/biz 维度 later 按需加,新增一个 `Strategy` 实现 + 注册进 Router.byName 即可,接口不变)。
|
||||
- 跨 region 故障转移(某 region 无账户时降级到其它 region)later。
|
||||
|
||||
---
|
||||
|
||||
## 后续阶段衔接
|
||||
|
||||
- **P6 对账 job**:实现读对账用量的 `accounts.UsageSource` 注入 `NewRouter`,limit_aware 即由内存空源切换到真实日累计,`LimitAware`/`Router` 代码不改。
|
||||
- **P3 crypto adapter**:在 `create()` 内按选中 account 的 xpub 派生地址(P5 已把 account 选对);地址池 = 对 crypto 渠道配 round_robin/weighted。
|
||||
Reference in New Issue
Block a user