wangjia
0ae7769abe
- gateway: TestSettleConcurrentSameOrder 真并发钉住 MarkAttemptPaid 行锁不变量 (文件型 sqlite,规避 in-memory cache=shared 的 SQLITE_LOCKED_SHAREDCACHE)。 - gateway: e2e_fullchain_test.go 补下单→回调→settle→webhook 实际 HTTP 投递→ delivered 整链(httptest server + 真实 store.WebhookStore/webhook.Notifier)。 - reconcile: main.go 装配抽到 reconcile.Assemble(+Runner.TaskNames 访问器), 补 assembly_test.go 钉住 7 个后台任务全部注册 + crypto 缺渠道/interval=0 分支。 - alipay: 补验签健壮性(未知多余字段/字段乱序/中文unicode/空值字段),意外定位 vendor Encoder 对空值字段的真实语义与官方文档描述不同,已记录在测试注释。 - money: 补 TestZeroDecimalCurrencyJPYNotSupported,记录 JPY/KRW 当前未注册进 exponents(接入会先踩 ErrUnknownCurrency,不是乘除法坑)。
117 lines
4.1 KiB
Go
117 lines
4.1 KiB
Go
// Package reconcile 是 pay v2 的后台守护/对账装配层:把多个幂等、崩溃安全、
|
|
// 可重跑的周期任务(订单过期清理 / webhook 死信硬化 / crypto 预留冷启动 /
|
|
// 用量刷新 / 查单对账 / 链上孤儿发现)挂到统一 Runner。它处于 main 之下的装配层,
|
|
// 可依赖 store/gateway/accounts/provider;gateway/provider 核心不反向依赖它。
|
|
package reconcile
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"time"
|
|
)
|
|
|
|
// Task 一个周期任务:名字 + 间隔 + 幂等可重跑的 Run。Local 标记该任务是否纯本地
|
|
// (只碰 DB,无出网 HTTP)——启动预热(RunOnceLocal)只跑 Local 任务,网络型任务交给
|
|
// 各自 ticker 首跳,避免上游慢拖住服务启动(main.go r.Run(addr) 之前的同步阶段)。
|
|
type Task struct {
|
|
Name string
|
|
Interval time.Duration
|
|
Local bool
|
|
Run func(ctx context.Context) error
|
|
}
|
|
|
|
// Runner 持有一组周期任务,逐个跑(RunOnce)或各自 ticker 常驻(Start)。
|
|
type Runner struct {
|
|
tasks []Task
|
|
logf func(format string, args ...any)
|
|
}
|
|
|
|
func NewRunner() *Runner { return &Runner{logf: log.Printf} }
|
|
|
|
// Add 注册一个周期任务(默认视为网络型/非 Local,启动预热 RunOnceLocal 不跑它,
|
|
// 交给 Start 里各自 ticker 的首跳执行)。
|
|
func (r *Runner) Add(name string, interval time.Duration, run func(ctx context.Context) error) {
|
|
r.add(Task{Name: name, Interval: interval, Local: false, Run: run})
|
|
}
|
|
|
|
// AddLocal 注册一个纯本地周期任务(只碰 DB,无出网 HTTP)——会被启动预热
|
|
// RunOnceLocal 同步执行一次,让 usage 快照/过期清理/退款自愈等立即生效。
|
|
func (r *Runner) AddLocal(name string, interval time.Duration, run func(ctx context.Context) error) {
|
|
r.add(Task{Name: name, Interval: interval, Local: true, Run: run})
|
|
}
|
|
|
|
// add 是 Add/AddLocal 的共同落地:Interval<=0 是 operator 配置错误(如
|
|
// expire_every_sec: 0)——time.NewTicker 对 <=0 的间隔会 panic,且 Start 里每 tick
|
|
// 的 recover 覆盖不到 NewTicker 本身(它在 goroutine 里、ticker 创建那一行就炸,
|
|
// 无 defer 保护)。显式优于静默改值:跳过注册 + WARN 日志,而不是偷偷 clamp 成默认值
|
|
// 掩盖配置错误。
|
|
func (r *Runner) add(t Task) {
|
|
if t.Interval <= 0 {
|
|
r.logf("[reconcile] WARN 任务 %s interval<=0(%v),跳过注册(检查配置)", t.Name, t.Interval)
|
|
return
|
|
}
|
|
r.tasks = append(r.tasks, t)
|
|
}
|
|
|
|
// exec 跑单个任务一次:panic recover + error 记录,绝不外抛(单任务失败不拖垮其它)。
|
|
func (r *Runner) exec(ctx context.Context, t Task) {
|
|
defer func() {
|
|
if rec := recover(); rec != nil {
|
|
r.logf("[reconcile] 任务 %s panic 已恢复: %v", t.Name, rec)
|
|
}
|
|
}()
|
|
if err := t.Run(ctx); err != nil {
|
|
r.logf("[reconcile] 任务 %s: %v", t.Name, err)
|
|
}
|
|
}
|
|
|
|
// RunOnce 顺序跑一遍全部任务(单测入口;main.go 启动预热改用 RunOnceLocal,
|
|
// 网络型任务不再阻塞启动,见下)。
|
|
func (r *Runner) RunOnce(ctx context.Context) {
|
|
for _, t := range r.tasks {
|
|
r.exec(ctx, t)
|
|
}
|
|
}
|
|
|
|
// RunOnceLocal 只顺序跑一遍 Local 任务(启动预热用):order-expire/usage-refresh/
|
|
// refund-apply-sweep/refund-stuck-alert 等纯 DB 任务立即生效;sync-pending/
|
|
// paid-spotcheck/crypto-orphan-scan 等有出网 HTTP(10-15s 超时)的任务跳过,交给
|
|
// Start 里各自 ticker 的首跳执行,避免上游慢拖住 r.Run(addr) 前的服务启动。
|
|
func (r *Runner) RunOnceLocal(ctx context.Context) {
|
|
for _, t := range r.tasks {
|
|
if !t.Local {
|
|
continue
|
|
}
|
|
r.exec(ctx, t)
|
|
}
|
|
}
|
|
|
|
// TaskNames 返回已注册任务名(注册序)。供装配期测试断言"预期任务集合是否都注册上"
|
|
// (见 assembly.go::Assemble 与 assembly_test.go),不用于运行期逻辑。
|
|
func (r *Runner) TaskNames() []string {
|
|
names := make([]string, len(r.tasks))
|
|
for i, t := range r.tasks {
|
|
names[i] = t.Name
|
|
}
|
|
return names
|
|
}
|
|
|
|
// Start 每任务一 goroutine + 独立 ticker 常驻;ctx 取消即退出。每 tick 崩溃安全。
|
|
func (r *Runner) Start(ctx context.Context) {
|
|
for _, t := range r.tasks {
|
|
t := t
|
|
go func() {
|
|
tk := time.NewTicker(t.Interval)
|
|
defer tk.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-tk.C:
|
|
r.exec(ctx, t)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
}
|