From 6427e8c8487b4be8df419fc441471fe42ce5047f Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Fri, 10 Jul 2026 17:19:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20reconcile=20Runner=20=E9=AA=A8?= =?UTF-8?q?=E6=9E=B6=20+=20=E8=AE=A2=E5=8D=95=E7=BA=A7=E8=BF=87=E6=9C=9F?= =?UTF-8?q?=E6=B8=85=E7=90=86(=E8=B6=85=20TTL=20pending=20=E5=8D=95/?= =?UTF-8?q?=E9=9B=B6=E5=B0=9D=E8=AF=95=E5=AD=A4=E5=84=BF=E5=8D=95=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E5=85=B3=E9=97=AD)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/reconcile/orders.go | 24 +++++++++++ internal/reconcile/orders_test.go | 31 ++++++++++++++ internal/reconcile/runner.go | 69 ++++++++++++++++++++++++++++++ internal/reconcile/runner_test.go | 23 ++++++++++ internal/store/order_query.go | 32 ++++++++++++++ internal/store/order_query_test.go | 45 +++++++++++++++++++ 6 files changed, 224 insertions(+) create mode 100644 internal/reconcile/orders.go create mode 100644 internal/reconcile/orders_test.go create mode 100644 internal/reconcile/runner.go create mode 100644 internal/reconcile/runner_test.go diff --git a/internal/reconcile/orders.go b/internal/reconcile/orders.go new file mode 100644 index 0000000..32aeaf4 --- /dev/null +++ b/internal/reconcile/orders.go @@ -0,0 +1,24 @@ +package reconcile + +import ( + "context" + "log" + "time" + + "github.com/wangjia/pay/internal/store" +) + +// OrderExpirerTask 返回「关闭超 TTL 未付 pending 订单」的周期任务体。 +// cutoff = now()-ttl,now 注入(测试确定性);幂等条件 UPDATE(见 store.ExpireStaleOrders)。 +func OrderExpirerTask(orders *store.OrderStore, ttl time.Duration, now func() time.Time) func(ctx context.Context) error { + return func(ctx context.Context) error { + n, err := orders.ExpireStaleOrders(now().Add(-ttl), 500) + if err != nil { + return err + } + if n > 0 { + log.Printf("[reconcile] 过期关闭 %d 个超时未付订单(TTL=%s)", n, ttl) + } + return nil + } +} diff --git a/internal/reconcile/orders_test.go b/internal/reconcile/orders_test.go new file mode 100644 index 0000000..76ae457 --- /dev/null +++ b/internal/reconcile/orders_test.go @@ -0,0 +1,31 @@ +package reconcile_test + +import ( + "context" + "testing" + "time" + + "github.com/wangjia/pay/internal/model" + "github.com/wangjia/pay/internal/reconcile" + "github.com/wangjia/pay/internal/store" +) + +func TestOrderExpirerTaskClosesStalePending(t *testing.T) { + db := model.OpenTestDB(t) + s := store.NewOrderStore(db) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + + o := &model.OrderV2{OutTradeNo: "OLD", AmountMinor: 100, Currency: "USDT", Status: model.OrderPendingV2} + _ = s.CreateOrder(o) + _ = db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "OLD"). + Update("created_at", now.Add(-2*time.Hour)).Error + + task := reconcile.OrderExpirerTask(s, time.Hour, func() time.Time { return now }) + if err := task(context.Background()); err != nil { + t.Fatalf("task: %v", err) + } + got, _ := s.GetOrder("OLD") + if got.Status != model.OrderExpiredV2 { + t.Fatalf("超时 pending 单应 expired, got %v", got.Status) + } +} diff --git a/internal/reconcile/runner.go b/internal/reconcile/runner.go new file mode 100644 index 0000000..817d3f1 --- /dev/null +++ b/internal/reconcile/runner.go @@ -0,0 +1,69 @@ +// Package reconcile 是 pay v2 的后台守护/对账装配层:把多个幂等、崩溃安全、 +// 可重跑的周期任务(订单过期清理 / webhook 死信硬化 / crypto 预留冷启动 / +// 用量刷新 / 查单对账 / 链上孤儿发现)挂到统一 Runner。它处于 main 之下的装配层, +// 可依赖 store/gateway/accounts/provider;gateway/provider 核心不反向依赖它。 +package reconcile + +import ( + "context" + "log" + "time" +) + +// Task 一个周期任务:名字 + 间隔 + 幂等可重跑的 Run。 +type Task struct { + Name string + Interval time.Duration + 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 注册一个周期任务。 +func (r *Runner) Add(name string, interval time.Duration, run func(ctx context.Context) error) { + r.tasks = append(r.tasks, Task{Name: name, Interval: interval, Run: run}) +} + +// 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 顺序跑一遍全部任务(启动预热 + 单测入口)。 +func (r *Runner) RunOnce(ctx context.Context) { + for _, t := range r.tasks { + r.exec(ctx, t) + } +} + +// 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) + } + } + }() + } +} diff --git a/internal/reconcile/runner_test.go b/internal/reconcile/runner_test.go new file mode 100644 index 0000000..0baa11f --- /dev/null +++ b/internal/reconcile/runner_test.go @@ -0,0 +1,23 @@ +package reconcile_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/wangjia/pay/internal/reconcile" +) + +func TestRunnerRunOnceExecutesAllAndRecoversPanic(t *testing.T) { + r := reconcile.NewRunner() + var a, b int + r.Add("inc-a", time.Minute, func(context.Context) error { a++; return nil }) + r.Add("boom", time.Minute, func(context.Context) error { panic("kaboom") }) // 不得拖垮后续 + r.Add("inc-b", time.Minute, func(context.Context) error { b++; return errors.New("soft") }) + + r.RunOnce(context.Background()) // panic 被 recover,error 被记录,均不中断 + if a != 1 || b != 1 { + t.Fatalf("a=%d b=%d, want 1/1(panic 任务不应阻断其它)", a, b) + } +} diff --git a/internal/store/order_query.go b/internal/store/order_query.go index 06867df..c306569 100644 --- a/internal/store/order_query.go +++ b/internal/store/order_query.go @@ -3,6 +3,7 @@ package store import ( "errors" "fmt" + "time" "gorm.io/gorm" @@ -94,3 +95,34 @@ func (s *OrderStore) ApplyRefundToOrder(outTradeNo string, fully bool) (bool, er } return res.RowsAffected > 0, nil } + +// ExpireStaleOrders closes pending orders whose created_at predates cutoff +// (TTL 到期未付),条件 UPDATE 只翻 status=pending 的行——与并发 settle 翻 paid +// 互斥(谁先谁赢,另一方 RowsAffected=0),故幂等且崩溃安全。含"零尝试孤儿单" +// (建单后 CreateAttempt 失败、无 attempt 的 pending 单):它 status 仍是 pending, +// 同样被扫到关闭(P2-T4 复审记录的缺口)。cutoff 由调用方用注入时钟算,store 不碰时钟。 +// +// 注:与 ExpirePendingAttempts 语义不同——那个是 attempt 级(标 attempt expired, +// 订单不动,retry 前用);这个是 order 级(标整张订单 expired)。 +func (s *OrderStore) ExpireStaleOrders(cutoff time.Time, limit int) (int64, error) { + if limit <= 0 || limit > 1000 { + limit = 500 + } + // 先选主键再批量 UPDATE:回避 "UPDATE ... ORDER BY LIMIT" 的方言差异(sqlite/mysql)。 + var ids []uint64 + if err := s.db.Model(&model.OrderV2{}). + Where("status = ? AND created_at < ?", model.OrderPendingV2, cutoff). + Order("id ASC").Limit(limit).Pluck("id", &ids).Error; err != nil { + return 0, fmt.Errorf("store.ExpireStaleOrders select: %w", err) + } + if len(ids) == 0 { + return 0, nil + } + res := s.db.Model(&model.OrderV2{}). + Where("id IN ? AND status = ?", ids, model.OrderPendingV2). // status 守卫兜住 select→update 间的并发翻转 + Update("status", model.OrderExpiredV2) + if res.Error != nil { + return 0, fmt.Errorf("store.ExpireStaleOrders update: %w", res.Error) + } + return res.RowsAffected, nil +} diff --git a/internal/store/order_query_test.go b/internal/store/order_query_test.go index a3f235c..40e509d 100644 --- a/internal/store/order_query_test.go +++ b/internal/store/order_query_test.go @@ -49,3 +49,48 @@ func TestListPendingAndExpire(t *testing.T) { } _ = time.Now } + +func TestExpireStaleOrders(t *testing.T) { + db := model.OpenTestDB(t) + s := store.NewOrderStore(db) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + + // 三张单:旧 pending(应过期)、新 pending(未到点)、旧 paid(不动)。 + mk := func(no string, st model.OrderStatusV2, createdAgo time.Duration) { + o := &model.OrderV2{OutTradeNo: no, AmountMinor: 100, Currency: "USDT", Status: st} + if err := s.CreateOrder(o); err != nil { + t.Fatalf("create %s: %v", no, err) + } + // 直接回填 created_at 制造账龄(GORM 默认写 now)。 + if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", no). + Update("created_at", now.Add(-createdAgo)).Error; err != nil { + t.Fatalf("backdate %s: %v", no, err) + } + } + mk("STALE", model.OrderPendingV2, 2*time.Hour) // 旧 pending + mk("FRESH", model.OrderPendingV2, 5*time.Minute) // 新 pending + mk("PAID", model.OrderPaidV2, 2*time.Hour) // 旧 paid + + cutoff := now.Add(-time.Hour) // TTL=1h + n, err := s.ExpireStaleOrders(cutoff, 500) + if err != nil { + t.Fatalf("expire: %v", err) + } + if n != 1 { + t.Fatalf("应关闭 1 张(仅 STALE), got %d", n) + } + assertStatus := func(no string, want model.OrderStatusV2) { + o, _ := s.GetOrder(no) + if o.Status != want { + t.Fatalf("%s status = %v, want %v", no, o.Status, want) + } + } + assertStatus("STALE", model.OrderExpiredV2) + assertStatus("FRESH", model.OrderPendingV2) + assertStatus("PAID", model.OrderPaidV2) + + // 幂等:再跑一次不再关闭(STALE 已非 pending)。 + if n2, _ := s.ExpireStaleOrders(cutoff, 500); n2 != 0 { + t.Fatalf("重跑应 0, got %d", n2) + } +}