feat(v2): crypto 预留表冷启动兜底——注入 loader 从 pending attempts 重建 reservation(兜重启丢内存)
This commit is contained in:
wangjia
@@ -49,15 +49,29 @@ type Provider struct {
|
||||
baseURL string
|
||||
http *http.Client
|
||||
now func() time.Time
|
||||
loader ReservationLoader // 冷启动预留重建源(装配期注入,nil=不重建)
|
||||
|
||||
mu sync.Mutex
|
||||
reserved map[string]time.Time // "<address>/<amount>" → 预留到期(链上匹配维度,对齐 Query 的 to==addr)(canonical AmountRecentlyUsed 的进程内等价)
|
||||
}
|
||||
|
||||
// PendingReservation 冷启动重建一笔预留所需的最小信息(中性结构,crypto 不依赖 store)。
|
||||
type PendingReservation struct {
|
||||
AccountID string // 收款账户(用于解析地址,链上匹配维度)
|
||||
AmountMinor int64 // attempt 冻结的 base 金额(不含尾数)
|
||||
ProviderRef string // "CRYPTO-<OutTradeNo>-<tail>",用于恢复尾数
|
||||
ReservedAt time.Time // 建单时间(= attempt.CreatedAt),冷却窗自此算
|
||||
}
|
||||
|
||||
// ReservationLoader 返回当前仍活跃(pending)的 crypto 预留。装配期由 main 用 OrderStore 实现。
|
||||
type ReservationLoader func(ctx context.Context) ([]PendingReservation, error)
|
||||
|
||||
type Option func(*Provider)
|
||||
|
||||
func WithBaseURL(u string) Option { return func(p *Provider) { p.baseURL = u } }
|
||||
func WithHTTPClient(c *http.Client) Option { return func(p *Provider) { p.http = c } }
|
||||
func WithBaseURL(u string) Option { return func(p *Provider) { p.baseURL = u } }
|
||||
func WithHTTPClient(c *http.Client) Option { return func(p *Provider) { p.http = c } }
|
||||
func WithReservationLoader(l ReservationLoader) Option { return func(p *Provider) { p.loader = l } }
|
||||
func WithNow(f func() time.Time) Option { return func(p *Provider) { p.now = f } }
|
||||
|
||||
func New(accts *accounts.Registry, opts ...Option) *Provider {
|
||||
p := &Provider{
|
||||
@@ -151,6 +165,40 @@ func tailFromRef(ref string) (int64, error) {
|
||||
return strconv.ParseInt(ref[i+1:], 10, 64)
|
||||
}
|
||||
|
||||
// Warm 冷启动兜底:把仍在冷却窗内的活跃预留灌回内存表,兜住重启丢 map 导致的金额复用误配。
|
||||
// 幂等:只加不覆盖更早到期时间;冷却已过的跳过。装配期在起服务前调一次即可。
|
||||
func (p *Provider) Warm(ctx context.Context) error {
|
||||
if p.loader == nil {
|
||||
return nil
|
||||
}
|
||||
items, err := p.loader(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
now := p.now()
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
for _, it := range items {
|
||||
addr, err := p.address(it.AccountID) // 地址是链上匹配维度真相源
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
tail, err := tailFromRef(it.ProviderRef)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
until := it.ReservedAt.Add(amountCooldown)
|
||||
if !until.After(now) {
|
||||
continue // 冷却已过,金额可安全复用,无需恢复
|
||||
}
|
||||
key := addr + "/" + strconv.FormatInt(it.AmountMinor+tail, 10)
|
||||
if cur, ok := p.reserved[key]; !ok || until.After(cur) {
|
||||
p.reserved[key] = until
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Provider) Create(_ context.Context, req provider.CreateRequest) (*provider.Session, error) {
|
||||
if req.Currency != "USDT" {
|
||||
return nil, fmt.Errorf("crypto: 仅支持 USDT, got %s", req.Currency)
|
||||
|
||||
@@ -206,3 +206,35 @@ func TestReservationKeyByAddress(t *testing.T) {
|
||||
t.Fatalf("共享地址的两个账户不应分配相同金额: %d", amt1)
|
||||
}
|
||||
}
|
||||
|
||||
// 冷启动兜底:装配期注入的 ReservationLoader 在 Warm 时把仍在冷却窗内的 pending
|
||||
// 预留灌回内存表;超冷却窗的旧预留(迟到旧款已不可能匹配)不必恢复。
|
||||
func TestWarmRebuildsReservationsFromLoader(t *testing.T) {
|
||||
const addr = "TWarmTestAddr000000000000000000000"
|
||||
t.Setenv("CRY_ADDRESS", addr)
|
||||
reg := accounts.New([]config.AccountConfig{
|
||||
{AccountID: "cry-1", Channel: "crypto", Enabled: true, CredentialEnvPrefix: "cry"},
|
||||
})
|
||||
now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
// 两条 pending:一条在冷却窗内(应恢复),一条建单于 40min 前(> 30min 冷却窗,应跳过)。
|
||||
loader := func(context.Context) ([]crypto.PendingReservation, error) {
|
||||
return []crypto.PendingReservation{
|
||||
{AccountID: "cry-1", AmountMinor: 29990000, ProviderRef: "CRYPTO-PAY-A-263", ReservedAt: now.Add(-5 * time.Minute)},
|
||||
{AccountID: "cry-1", AmountMinor: 29990000, ProviderRef: "CRYPTO-PAY-B-777", ReservedAt: now.Add(-40 * time.Minute)},
|
||||
}, nil
|
||||
}
|
||||
p := crypto.New(reg, crypto.WithReservationLoader(loader), crypto.WithNow(func() time.Time { return now }))
|
||||
if err := p.Warm(context.Background()); err != nil {
|
||||
t.Fatalf("warm: %v", err)
|
||||
}
|
||||
res := p.GetReserved()
|
||||
inWindow := addr + "/" + strconv.FormatInt(29990000+263, 10)
|
||||
expired := addr + "/" + strconv.FormatInt(29990000+777, 10)
|
||||
if _, ok := res[inWindow]; !ok {
|
||||
t.Fatalf("冷却窗内的预留应恢复, got %v", res)
|
||||
}
|
||||
if _, ok := res[expired]; ok {
|
||||
t.Fatalf("超冷却窗的预留不应恢复, got %v", res)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
package reconcile
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/provider/crypto"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
// CryptoReservationLoader 装配 crypto 冷启动预留源:读全部 pending attempt,
|
||||
// 过滤 channel=crypto,映射为 crypto.PendingReservation。ReservedAt 取 attempt.CreatedAt
|
||||
// (建单时刻,冷却窗自此算)。crypto 不 import store,故此桥在装配层。
|
||||
func CryptoReservationLoader(orders *store.OrderStore) crypto.ReservationLoader {
|
||||
return func(ctx context.Context) ([]crypto.PendingReservation, error) {
|
||||
atts, err := orders.ListAttemptsByStatus(model.AttemptPending, 200)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]crypto.PendingReservation, 0, len(atts))
|
||||
for i := range atts {
|
||||
a := &atts[i]
|
||||
if a.Channel != "crypto" {
|
||||
continue
|
||||
}
|
||||
out = append(out, crypto.PendingReservation{
|
||||
AccountID: a.AccountID, AmountMinor: a.AmountMinor,
|
||||
ProviderRef: a.ProviderRef, ReservedAt: a.CreatedAt,
|
||||
})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package reconcile_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/wangjia/pay/internal/model"
|
||||
"github.com/wangjia/pay/internal/reconcile"
|
||||
"github.com/wangjia/pay/internal/store"
|
||||
)
|
||||
|
||||
func TestCryptoReservationLoaderFiltersPendingCrypto(t *testing.T) {
|
||||
db := model.OpenTestDB(t)
|
||||
s := store.NewOrderStore(db)
|
||||
// 两条 attempt:crypto pending(要)、alipay pending(不要)。
|
||||
_ = s.CreateAttempt(&model.Attempt{OutTradeNo: "O1", Channel: "crypto", ProviderRef: "CRYPTO-O1-12",
|
||||
AmountMinor: 100, Currency: "USDT", Status: model.AttemptPending})
|
||||
_ = s.CreateAttempt(&model.Attempt{OutTradeNo: "O2", Channel: "alipay", ProviderRef: "AL-O2",
|
||||
AmountMinor: 200, Currency: "CNY", Status: model.AttemptPending})
|
||||
|
||||
loader := reconcile.CryptoReservationLoader(s)
|
||||
items, err := loader(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("loader: %v", err)
|
||||
}
|
||||
if len(items) != 1 || items[0].ProviderRef != "CRYPTO-O1-12" || items[0].AmountMinor != 100 {
|
||||
t.Fatalf("只应含 crypto pending, got %+v", items)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user