pay/docs/superpowers/plans/2026-07-10-pay-v2-p6-reconcile.md

pay v2 · P6 对账与后台守护 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. 每个 Task 自包含,给完整 Go 代码 + 测试 + 命令,可交给全新 implementer subagent。

设计文档(全景蓝图): docs/pay-v2-unified-gateway-design.html。P6 直接兑现其 §8 结尾铁律「对账优先于实时:每 provider 实现 query,跑定时对账 job 防掉单/防重复授权」。 前置计划:

• docs/superpowers/plans/2026-07-10-pay-v2-p1-core-model.md(P1:model.OrderV2/Attempt/WebhookDelivery、store.OrderStore/WebhookStore、accounts)。
• docs/superpowers/plans/2026-07-10-pay-v2-p2-pipeline.md(P2:gateway.Settle/SyncPendingAttempts、webhook outbox、store.ExpirePendingAttempts;本计划直接复用其接口,不重造)。P2 计划 Self-Review「后续阶段」条目明确把订单级过期清理与 Notifier 硬化(退避/死信/告警) routed 到 P6。
• docs/superpowers/plans/2026-07-10-pay-v2-p3-channel-adapters.md(P3:crypto 自托管 adapter + 进程内预留表)。P3 Self-Review「已知取舍」把 crypto 预留表持久化冷启动兜底 与 crypto orphan 到账落表 routed 到 P6。
• docs/superpowers/plans/2026-07-10-pay-v2-p5-account-routing.md(P5:accounts.Router/LimitAware/UsageSource)。P5 计划 D5 明确 LimitAware 真实用量数据源由 P6 对账 job 提供,届时实现 accounts.UsageSource 注入 Router 即可,LimitAware/Router 不改。

Goal: 给 pay v2 装上一套周期后台守护 + 对账:统一 internal/reconcile 包挂多个幂等、崩溃安全、可重跑的周期任务——① 订单级过期清理(超 TTL 未付 pending 单自动关闭,含零尝试孤儿单);② Notifier 投递硬化(指数退避 + 最大次数 + 死信标记 + 告警钩子);③ crypto 预留表持久化冷启动兜底(用 attempts 表重建 reservation,兜住重启丢内存);④ LimitAware 真实用量数据源(attempts 按账户结算币种聚合当日已收);⑤ 对账主体(周期 SyncPendingAttempts + 已付订单抽查核对);⑥ crypto 孤儿链上支付发现(到账但不匹配任何 attempt → 落表 + 告警)。全程注入时钟 + 假 provider/httptest,免 docker、不打真网。

Architecture: 后台 job 统一形态 = reconcile.Runner:注册多个 Task{Name, Interval, Run(ctx)},Start(ctx) 每任务一 goroutine + 独立 ticker + recover(单 tick panic 不拖垮其它),RunOnce(ctx) 顺序跑一遍(启动预热 + 单测入口)。每个 Task 是幂等 sweep(条件 UPDATE / 快照刷新 / 查单收敛),重跑安全、崩溃后下一 tick 自愈。reconcile 处于装配层(可 import store/gateway/accounts/provider,类比 main),而 gateway/provider 核心仍保持中性——不反向依赖 reconcile。过期清理走 store 新增条件 UPDATE;Notifier 硬化改 webhook + store.WebhookStore + model.WebhookDelivery 加列;crypto 冷启动兜底经注入 ReservationLoader(crypto 不 import store,装配期 main 提供 loader);用量数据源 reconcile.UsageSource 结构体满足 accounts.UsageSource 接口(Router 注入它、刷新 job 也持它);orphan 经中性可选接口 provider.OrphanScanner(crypto 实现,reconcile type-assert)。

Tech Stack: Go 1.26.1 · github.com/wangjia/pay · Gin · GORM v1.31 · glebarez/sqlite · 标准库 time/context/sync/log。无新增第三方依赖。crypto orphan 复用 P3 已有 TronGrid REST 客户端(p.http + only_confirmed)。

Global Constraints(继承 P2/P3/P5)

• 复用不重造:模型 model.OrderV2/Attempt/WebhookDelivery + V2 后缀状态;数据访问 store.OrderStore/WebhookStore;管线 gateway.Gateway(Settle/SyncPendingAttempts);账户 accounts.Registry/Router/LimitAware/UsageSource(接口)/config.AccountConfig;Provider 抽象 provider.Provider/QueryRequest/PaidEvent;crypto adapter internal/provider/crypto(allocateAmount/Query/tailFromRef/address/进程内 reserved map)。
• 金额一律 int64 最小单位 + 币种码,禁 float;时间一律 Go 端算好传 ?(cutoff/since 由 job 用注入时钟算,不写 SQL 时间函数,保持 sqlite/mysql 可移植)。
• 幂等 + 崩溃安全:每个 job 是可重跑 sweep;订单过期用 status = pending 守卫的条件 UPDATE(与并发 settle 翻转不打架:谁先谁赢,另一方 RowsAffected=0 no-op);死信/退避靠列状态持久化,重启续跑;usage 快照刷新是幂等覆盖;orphan 落表按 tx_id unique 去重。
• Provider 中性守恒:internal/gateway、internal/provider(核心 provider.go)不 import reconcile;crypto 不 import store(经注入 loader / 中性接口传数据)。reconcile 是装配层,只被 main 调。
• 凭证 env only,测试用 t.Setenv;所有测试 :memory:/临时 sqlite + httptest 假 TronGrid/假业务方,免 docker、不打真网。
• 每步 go build ./... 通过;测试 go test ./... 全绿。每任务严格 bite-sized TDD:写失败测试 → 跑失败 → 实现 → 跑通过 → commit。禁占位。

债务台账落点(prior reviews routed 到 P6 的 6 项 + 1 已做项)

#	债务项	落点
1	订单级过期清理(超 TTL pending 单关闭,含零尝试孤儿 pending 单)	Task 1
2	Notifier 重试硬化(退避 / 最大次数 / 死信 / 告警钩子)	Task 2
3	crypto 孤儿链上支付(到账不匹配任何 attempt → 记录 + 告警)	Task 6
4	crypto 预留表多实例/重启兜底(attempts 表冷启动重建)	Task 3
5	LimitAware 真实数据源(attempts/orders 按账户结算币种聚合)	Task 4
6	对账主体(周期 SyncPendingAttempts + 渠道流水 vs 本地订单核对)	Task 5(决策:全量对账单下载判定过重 → 降级为「逐 pending attempt query + 已付订单抽查」,见 Task 5 Self-Review)
7	settle-sync 非常态结果日志(764ed55 已做)	已存在,不重复;Task 5 调度 SyncPendingAttempts 时天然继承其日志

---

Task 1: reconcile.Runner 骨架 + 订单级过期清理(ExpireStaleOrders)

债务 #1。 建后台 job 统一形态(Runner),并落地第一个 job:超 TTL 未付 pending 订单自动关闭。含 P2-T4 记录的「零尝试孤儿 pending 单」——建单后 prov.Create/CreateAttempt 失败留下的、调用方拿不到单号无法 retry/cancel 的 pending 单,因其 status=pending 同样被扫到。注意与 attempt 级过期区分:P2 已有 store.ExpirePendingAttempts(把某单的 pending attempt 标 expired,订单状态不动,retry 前调用);本任务是订单级——把整张 pending 订单标 expired。

Files:

• Create: internal/reconcile/runner.go(Runner 骨架)
• Create: internal/reconcile/runner_test.go
• Create: internal/reconcile/orders.go(OrderExpirerTask 工厂)
• Create: internal/reconcile/orders_test.go
• Modify: internal/store/order_query.go(新增 ExpireStaleOrders)
• Modify: internal/store/order_query_test.go(若无则新建,断言 ExpireStaleOrders)

Interfaces:

• store.(*OrderStore).ExpireStaleOrders(cutoff time.Time, limit int) (int64, error) —— 条件 UPDATE:status=pending AND created_at < cutoff → expired;返回关闭条数。cutoff 由调用方(job)用注入时钟算(now-TTL),store 不碰时钟。

• reconcile.Task{Name string; Interval time.Duration; Run func(ctx) error};(*Runner).Add(name, interval, run) / RunOnce(ctx) / Start(ctx)。

• Step 1: 写失败测试

internal/store/order_query_test.go 追加(若文件不存在则新建,package store 内部测试以便直接改 created_at):

package store

import (
	"testing"
	"time"

	"github.com/wangjia/pay/internal/model"
)

func TestExpireStaleOrders(t *testing.T) {
	db := model.OpenTestDB(t)
	s := 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)
	}
}

internal/reconcile/runner_test.go:

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)
	}
}

internal/reconcile/orders_test.go:

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)
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/reconcile/ 2>&1 | tail -20 Expected: 编译失败——ExpireStaleOrders / reconcile 包不存在。

• Step 3: 写实现

internal/store/order_query.go 追加(import "time" 已需要;文件顶已 import errors/fmt/gorm/model,补 "time"):

// 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
}

internal/reconcile/runner.go:

// 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)
				}
			}
		}()
	}
}

internal/reconcile/orders.go:

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
	}
}

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go build ./... && go test ./internal/store/ ./internal/reconcile/ -v 2>&1 | tail -30 Expected: 全 PASS。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/reconcile/runner.go internal/reconcile/runner_test.go internal/reconcile/orders.go internal/reconcile/orders_test.go internal/store/order_query.go internal/store/order_query_test.go
git commit -m "feat(v2): reconcile Runner 骨架 + 订单级过期清理(超 TTL pending 单/零尝试孤儿单自动关闭)"

---

Task 2: Notifier 投递硬化 —— 指数退避 + 最大次数 + 死信标记 + 告警钩子

债务 #2。 现状 webhook/notifier.go 的 Start 是裸 60s ticker、MarkFailed 只 attempts+1——无退避(业务方挂了每 60s 猛敲)、无上限(永远重投)、无死信/告警(投不出去的单静默堆积)。本任务:失败走指数退避(next_attempt_at = now + base·2^(attempts-1),封顶 maxBackoff),达 maxAttempts 标 dead=true 并触发告警钩子;DeliverPending 只取「未投递、未死信、且 next_attempt_at 到点」的行。不破坏现有幂等(unique 键入队)与投递门禁(订单未付不投、不计失败)。

Files:

• Modify: internal/model/webhook_delivery.go(加 Dead + NextAttemptAt 列)
• Modify: internal/store/webhook.go(ListDeliverable / ScheduleRetry / MarkDead)
• Modify: internal/store/webhook_test.go(断言新查询/调度)
• Modify: internal/webhook/notifier.go(functional options:时钟/上限/退避/告警;失败分流退避 vs 死信)
• Modify: internal/webhook/notifier_test.go(注入时钟测退避/死信/告警;修 TestNotifierRetriesOnFailure 让第二次投递前推进时钟越过退避)

Interfaces:

• store.(*WebhookStore).ListDeliverable(now time.Time, limit int) ([]WebhookDeliveryRow, error)
• store.(*WebhookStore).ScheduleRetry(id uint64, errMsg string, nextAt time.Time) error(attempts+1、last_error、next_attempt_at)
• store.(*WebhookStore).MarkDead(id uint64, errMsg string) error(attempts+1、dead=true、last_error)
• webhook.NewNotifier(ws, bizConfig, orderPaid, opts ...Option) + WithClock / WithMaxAttempts / WithBaseBackoff / WithMaxBackoff / WithAlerter。保留原 3 参调用式(options 变参),现有 main/测试不改签名即可编译。

AutoMigrate 只加不删,新列对存量库安全(Dead 默认 false,NextAttemptAt NULL = 立即可投)。

• Step 1: 写失败测试

internal/store/webhook_test.go 追加:

func TestWebhookScheduleRetryAndDead(t *testing.T) {
	ws := store.NewWebhookStore(model.OpenTestDB(t))
	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	_ = ws.EnqueueDelivery("PAY-1", "pangolin", "payment.succeeded", `{"x":1}`)

	// 刚入队:next_attempt_at NULL → 立即可投。
	rows, _ := ws.ListDeliverable(now, 10)
	if len(rows) != 1 {
		t.Fatalf("新单应可投, got %d", len(rows))
	}
	id := rows[0].ID

	// 排下一次重试到 now+30s:此刻不可投,过点可投。
	if err := ws.ScheduleRetry(id, "http 500", now.Add(30*time.Second)); err != nil {
		t.Fatalf("schedule: %v", err)
	}
	if r, _ := ws.ListDeliverable(now, 10); len(r) != 0 {
		t.Fatalf("退避窗内不应可投, got %d", len(r))
	}
	if r, _ := ws.ListDeliverable(now.Add(31*time.Second), 10); len(r) != 1 || r[0].Attempts != 1 {
		t.Fatalf("过退避点应可投且 attempts=1, got %+v", r)
	}

	// 标死信:不再出现在可投集。
	if err := ws.MarkDead(id, "gave up"); err != nil {
		t.Fatalf("markdead: %v", err)
	}
	if r, _ := ws.ListDeliverable(now.Add(time.Hour), 10); len(r) != 0 {
		t.Fatalf("死信不应可投, got %d", len(r))
	}
}

internal/webhook/notifier_test.go —— 新增退避/死信/告警用例,并改造 TestNotifierRetriesOnFailure(退避后需推进时钟才重试):

// 业务方持续 500:每次失败按指数退避重排;时钟推进后才重投;达上限标死信 + 告警。
func TestNotifierBackoffThenDeadWithAlert(t *testing.T) {
	var hits int
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		hits++
		w.WriteHeader(http.StatusInternalServerError)
	}))
	defer srv.Close()

	clk := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	nowFn := func() time.Time { return clk }
	var alerted []string
	ws := store.NewWebhookStore(model.OpenTestDB(t))
	n := webhook.NewNotifier(ws,
		func(string) (config.BizSystemConfig, bool) {
			return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true
		},
		func(string) (bool, error) { return true, nil },
		webhook.WithClock(nowFn),
		webhook.WithBaseBackoff(time.Second),
		webhook.WithMaxBackoff(4*time.Second),
		webhook.WithMaxAttempts(3),
		webhook.WithAlerter(func(d *store.WebhookDeliveryRow, reason string) { alerted = append(alerted, d.OutTradeNo) }),
	)
	_ = n.Enqueue("PAY-D", "pangolin", "payment.succeeded", map[string]any{"event_type": "payment.succeeded"})

	// 尝试 1:失败 → attempts=1,退避到 +1s。
	if sent, _ := n.DeliverPending(10); sent != 0 || hits != 1 {
		t.Fatalf("try1 sent=%d hits=%d", sent, hits)
	}
	// 退避窗内不投。
	if sent, _ := n.DeliverPending(10); sent != 0 || hits != 1 {
		t.Fatalf("退避窗内不应再敲, hits=%d", hits)
	}
	// 推进越过退避;尝试 2 失败 → attempts=2,退避到 +2s。
	clk = clk.Add(2 * time.Second)
	if sent, _ := n.DeliverPending(10); sent != 0 || hits != 2 {
		t.Fatalf("try2 hits=%d", hits)
	}
	// 推进;尝试 3 失败 → attempts 达 maxAttempts(3)→ 死信 + 告警。
	clk = clk.Add(4 * time.Second)
	if sent, _ := n.DeliverPending(10); sent != 0 || hits != 3 {
		t.Fatalf("try3 hits=%d", hits)
	}
	if len(alerted) != 1 || alerted[0] != "PAY-D" {
		t.Fatalf("死信应触发告警一次, got %+v", alerted)
	}
	// 已死信:无论时钟怎么走都不再投。
	clk = clk.Add(time.Hour)
	if sent, _ := n.DeliverPending(10); sent != 0 || hits != 3 {
		t.Fatalf("死信后不应再投, hits=%d", hits)
	}
}

并把原 TestNotifierRetriesOnFailure 里第二次 DeliverPending 前插入时钟推进 —— 该测试改为注入时钟:

func TestNotifierRetriesOnFailure(t *testing.T) {
	var hits int
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		hits++
		w.WriteHeader(http.StatusInternalServerError)
	}))
	defer srv.Close()

	clk := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	ws := store.NewWebhookStore(model.OpenTestDB(t))
	n := webhook.NewNotifier(ws, func(string) (config.BizSystemConfig, bool) {
		return config.BizSystemConfig{CallbackURL: srv.URL, Secret: "x"}, true
	}, func(string) (bool, error) { return true, nil },
		webhook.WithClock(func() time.Time { return clk }),
		webhook.WithBaseBackoff(time.Second))
	_ = n.Enqueue("PAY-3", "pangolin", "payment.succeeded", map[string]any{"event_type": "payment.succeeded"})

	if sent, _ := n.DeliverPending(10); sent != 0 {
		t.Fatalf("失败不应算投递成功, got %d", sent)
	}
	pend, _ := ws.ListUndelivered(10)
	if len(pend) != 1 || pend[0].Attempts != 1 {
		t.Fatalf("失败后应留队重试, got %+v", pend)
	}
	clk = clk.Add(2 * time.Second) // 越过退避窗
	if _, _ = n.DeliverPending(10); hits < 2 {
		t.Fatalf("应重试第二次, hits=%d", hits)
	}
}

time 需加入 notifier_test.go 的 import。ws.ListUndelivered 保留(列 delivered=false 全量,供该断言),Notifier 内部改用 ListDeliverable。

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/webhook/ 2>&1 | tail -20 Expected: 编译失败——ListDeliverable/ScheduleRetry/MarkDead/WithClock 等未定义;Dead/NextAttemptAt 列不存在。

• Step 3: 写实现

internal/model/webhook_delivery.go —— 结构体尾部加两列:

	Delivered  bool   `gorm:"index;default:false" json:"delivered"`
	Attempts   int    `json:"attempts"`
	LastError  string `gorm:"size:255" json:"last_error,omitempty"`
	Dead          bool       `gorm:"index;default:false" json:"dead"`                 // 达最大次数放弃投递(死信),需人工/对账介入
	NextAttemptAt *time.Time `gorm:"index" json:"next_attempt_at,omitempty"`          // 指数退避的下次可投时刻;nil=立即可投

文件加 import "time"。

internal/store/webhook.go —— 新增三方法,并保留 ListUndelivered/MarkDelivered(MarkFailed 可删,已无调用方;保留亦无害,这里替换为下面两法):

// ListDeliverable 取「未投递、未死信、且退避到点(next_attempt_at NULL 或 <= now)」的行。
func (s *WebhookStore) ListDeliverable(now time.Time, limit int) ([]WebhookDeliveryRow, error) {
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	var out []WebhookDeliveryRow
	if err := s.db.
		Where("delivered = ? AND dead = ? AND (next_attempt_at IS NULL OR next_attempt_at <= ?)", false, false, now).
		Order("id ASC").Limit(limit).Find(&out).Error; err != nil {
		return nil, fmt.Errorf("store.ListDeliverable: %w", err)
	}
	return out, nil
}

// ScheduleRetry 记一次失败并排下次重试:attempts+1、last_error、next_attempt_at=nextAt。
func (s *WebhookStore) ScheduleRetry(id uint64, errMsg string, nextAt time.Time) error {
	errMsg = truncateUTF8(errMsg, 255)
	if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
		Updates(map[string]any{
			"attempts":        gorm.Expr("attempts + 1"),
			"last_error":      errMsg,
			"next_attempt_at": nextAt,
		}).Error; err != nil {
		return fmt.Errorf("store.ScheduleRetry: %w", err)
	}
	return nil
}

// MarkDead 达最大次数后放弃:attempts+1、dead=true、last_error。行留库供人工/对账排查。
func (s *WebhookStore) MarkDead(id uint64, errMsg string) error {
	errMsg = truncateUTF8(errMsg, 255)
	if err := s.db.Model(&model.WebhookDelivery{}).Where("id = ?", id).
		Updates(map[string]any{
			"attempts":   gorm.Expr("attempts + 1"),
			"dead":       true,
			"last_error": errMsg,
		}).Error; err != nil {
		return fmt.Errorf("store.MarkDead: %w", err)
	}
	return nil
}

import "time" 加入 webhook.go(store 包)。若删 MarkFailed,确认无其它调用方(grep -rn MarkFailed internal/,当前仅 notifier 用,Step 3 一并改掉)。

internal/webhook/notifier.go —— 加 options + 失败分流:

// 结构体加字段
type Notifier struct {
	deliveries  *store.WebhookStore
	bizConfig   BizConfigFunc
	orderPaid   OrderPaidFunc
	client      *http.Client
	now         func() time.Time
	maxAttempts int
	baseBackoff time.Duration
	maxBackoff  time.Duration
	alert       func(d *store.WebhookDeliveryRow, reason string)
}

type Option func(*Notifier)

func WithClock(f func() time.Time) Option           { return func(n *Notifier) { n.now = f } }
func WithMaxAttempts(m int) Option                  { return func(n *Notifier) { n.maxAttempts = m } }
func WithBaseBackoff(d time.Duration) Option        { return func(n *Notifier) { n.baseBackoff = d } }
func WithMaxBackoff(d time.Duration) Option         { return func(n *Notifier) { n.maxBackoff = d } }
func WithAlerter(a func(d *store.WebhookDeliveryRow, reason string)) Option {
	return func(n *Notifier) { n.alert = a }
}

func NewNotifier(ws *store.WebhookStore, bizConfig BizConfigFunc, orderPaid OrderPaidFunc, opts ...Option) *Notifier {
	n := &Notifier{
		deliveries: ws, bizConfig: bizConfig, orderPaid: orderPaid,
		client:      &http.Client{Timeout: 10 * time.Second},
		now:         time.Now,
		maxAttempts: 12,               // ~ 覆盖数小时退避后放弃(见 backoffFor 封顶)
		baseBackoff: 30 * time.Second, // 首次失败退避基
		maxBackoff:  time.Hour,        // 单次退避封顶
	}
	for _, o := range opts {
		o(n)
	}
	if n.alert == nil {
		n.alert = func(d *store.WebhookDeliveryRow, reason string) {
			log.Printf("[webhook][死信] out_trade_no=%s biz=%s event=%s attempts=%d 放弃投递: %s",
				d.OutTradeNo, d.BizSystem, d.EventType, d.Attempts, reason)
		}
	}
	return n
}

// backoffFor 计算第 attempts 次失败后的退避:base·2^(attempts-1),封顶 maxBackoff。
// attempts≥1;移位用 uint 且封顶,防溢出。
func (n *Notifier) backoffFor(attempts int) time.Duration {
	d := n.baseBackoff
	for i := 1; i < attempts; i++ {
		d *= 2
		if d >= n.maxBackoff {
			return n.maxBackoff
		}
	}
	if d > n.maxBackoff {
		return n.maxBackoff
	}
	return d
}

// fail 统一失败分流:未达上限→退避重排;达上限→死信+告警。
func (n *Notifier) fail(d *store.WebhookDeliveryRow, msg string) {
	attempts := d.Attempts + 1 // 本次即将记的失败次数
	if attempts >= n.maxAttempts {
		_ = n.deliveries.MarkDead(d.ID, msg)
		n.alert(d, msg)
		return
	}
	_ = n.deliveries.ScheduleRetry(d.ID, msg, n.now().Add(n.backoffFor(attempts)))
}

DeliverPending 改用 ListDeliverable(n.now(), limit):

func (n *Notifier) DeliverPending(limit int) (int, error) {
	rows, err := n.deliveries.ListDeliverable(n.now(), limit)
	if err != nil {
		return 0, err
	}
	ok := 0
	for i := range rows {
		if n.deliverOne(&rows[i]) {
			ok++
		}
	}
	return ok, nil
}

deliverOne 把原先所有 n.deliveries.MarkFailed(...) 调用改成 n.fail(d, <msg>);投递门禁(订单未付)分支不变(仍 return false 不计失败);成功仍 MarkDelivered。逐处替换:

	// biz 未配置
	if !found || cfg.CallbackURL == "" {
		n.fail(d, "biz system not configured")
		return false
	}
	...
	// 构建请求失败 / Do 失败 / 非 200 或无 SUCCESS
	n.fail(d, err.Error())          // 三处原 MarkFailed → fail
	...
	n.fail(d, fmt.Sprintf("http %d: %s", resp.StatusCode, truncate(string(rb), 120)))

Start 保持 60s ticker 不变(现在每轮 DeliverPending 已退避感知);其 log 文案可留。

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go build ./... && go test ./internal/store/ ./internal/webhook/ -v 2>&1 | tail -30 Expected: 全 PASS(含改造后的 retry 用例 + 新死信/告警用例)。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/model/webhook_delivery.go internal/store/webhook.go internal/store/webhook_test.go internal/webhook/notifier.go internal/webhook/notifier_test.go
git commit -m "feat(v2): webhook 投递硬化——指数退避 + 最大次数死信 + 告警钩子(替裸 60s 猛敲)"

---

Task 3: crypto 预留表持久化冷启动兜底(attempts 表重建 reservation)

债务 #4。 P3 crypto 的唯一金额预留表是进程内 reserved map——重启即丢,重启后并发同价新单可能复用一个仍在冷却窗内的金额,让迟到旧款误配新单。canonical(pangolin-pay)用自库 AmountRecentlyUsed 查重;pay 移植方案(P3 复审记录):用 attempts 表冷启动重建。本任务给 crypto 注入一个 ReservationLoader(装配期 main 提供,读 pending crypto attempts),Warm(ctx) 时把每个未过冷却窗的 (地址,期望金额) 灌回 reserved。crypto 不 import store —— loader 是注入函数,数据经中性结构传入,tail→期望金额的换算仍封在 crypto 内(复用 tailFromRef)。

Files:

• Modify: internal/provider/crypto/crypto.go(ReservationLoader 类型 + WithReservationLoader/WithNow Option + Warm)
• Modify: internal/provider/crypto/crypto_test.go(Warm 重建 + 冷却过期跳过)
• Create: internal/reconcile/crypto_warm.go(装配期 loader 工厂:从 OrderStore 读 pending crypto attempts)
• Create: internal/reconcile/crypto_warm_test.go

Interfaces:

• crypto.PendingReservation{AccountID string; AmountMinor int64; ProviderRef string; ReservedAt time.Time}(AmountMinor=attempt 冻结的 base 金额,ProviderRef=CRYPTO-<no>-<tail> 供恢复尾数,ReservedAt=建单时间=attempt.CreatedAt,冷却窗从此算)。

• crypto.ReservationLoader func(ctx context.Context) ([]PendingReservation, error);crypto.WithReservationLoader(l);crypto.WithNow(f)(测试注入时钟)。

• crypto.(*Provider).Warm(ctx) error。

• reconcile.CryptoReservationLoader(orders *store.OrderStore) crypto.ReservationLoader(读 ListAttemptsByStatus(AttemptPending,…) 过滤 Channel=="crypto",映射为 PendingReservation)。

• Step 1: 写失败测试

internal/provider/crypto/crypto_test.go 追加(复用文件里既有的 t.Setenv("CRY_ADDRESS",…) + accounts.New 模式;GetReserved 已由 export_test.go 提供):

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)
	}
}

import 补 context/strconv/time(crypto_test.go 若未 import)。tail 263/777 与 base 29990000 相加即期望金额,键式 <addr>/<amount> 与 allocateAmount 一致。

internal/reconcile/crypto_warm_test.go:

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)
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/provider/crypto/ ./internal/reconcile/ 2>&1 | tail -20 Expected: 编译失败——PendingReservation/WithReservationLoader/WithNow/Warm/CryptoReservationLoader 未定义。

• Step 3: 写实现

internal/provider/crypto/crypto.go —— 结构体加 loader,补 Option 与 Warm:

// 结构体加一行:
type Provider struct {
	accts    *accounts.Registry
	baseURL  string
	http     *http.Client
	now      func() time.Time
	loader   ReservationLoader // 冷启动预留重建源(装配期注入,nil=不重建)
	mu       sync.Mutex
	reserved map[string]time.Time
}

// 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)

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 } }

// 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
}

New 里 now: time.Now 默认已存在,WithNow 允许测试覆盖。import "context" crypto.go 已有。

internal/reconcile/crypto_warm.go:

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
	}
}

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go build ./... && go test ./internal/provider/crypto/ ./internal/reconcile/ -v 2>&1 | tail -30 Expected: 全 PASS。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/provider/crypto/crypto.go internal/provider/crypto/crypto_test.go internal/reconcile/crypto_warm.go internal/reconcile/crypto_warm_test.go
git commit -m "feat(v2): crypto 预留表冷启动兜底——注入 loader 从 pending attempts 重建 reservation(兜重启丢内存)"

---

Task 4: LimitAware 真实用量数据源(DBUsageSource + RefreshUsage job)

债务 #5。 P5 的 accounts.LimitAware 靠 UsageSource.TodayUsedMinor(accountID) 判越限,生产装的是 NopUsage(恒 0,退化为 round_robin)。P5 D5 约定:真实数据源由 P6 提供。本任务实现 reconcile.UsageSource(满足 accounts.UsageSource 接口):持一个「账户→当日已收 minor」快照,由对账 job 周期 Refresh 从 attempts 表聚合(status=paid AND paid_at>=当日起点,GROUP BY account_id 求 SUM(amount_minor))。量纲:attempt.AmountMinor 即该账户所属渠道结算币种的 minor(config 注释已在),故按 account_id 直接求和与 DailyLimit 同量纲,可比。快照读写加锁,Pick 路径零 DB(读快照);启动预热 Refresh 一次,之后 job 刷。

Files:

• Modify: internal/store/order_query.go(SumPaidAttemptMinorByAccountSince)
• Modify: internal/store/order_query_test.go(断言聚合)
• Create: internal/reconcile/usage.go(UsageSource 结构体 + Refresh + RefreshUsageTask + StartOfDay)
• Create: internal/reconcile/usage_test.go

Interfaces:

• store.(*OrderStore).SumPaidAttemptMinorByAccountSince(since time.Time) (map[string]int64, error)

• reconcile.NewUsageSource(orders *store.OrderStore, now func() time.Time) *UsageSource;(*UsageSource).TodayUsedMinor(accountID string) int64(满足 accounts.UsageSource);(*UsageSource).Refresh(ctx) error;reconcile.RefreshUsageTask(u *UsageSource) func(ctx) error。

• 当日起点用 now() 的日期(UTC)算(StartOfDay)——与 DailyLimit「按自然日」语义一致;跨区多实例各自 UTC 日窗,单实例部署无歧义。

• Step 1: 写失败测试

internal/store/order_query_test.go 追加:

func TestSumPaidAttemptMinorByAccountSince(t *testing.T) {
	db := model.OpenTestDB(t)
	s := NewOrderStore(db)
	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	dayStart := time.Date(2026, 7, 10, 0, 0, 0, 0, time.UTC)

	mk := func(no, acct string, minor int64, st model.AttemptStatus, paidAgo time.Duration) {
		paid := now.Add(-paidAgo)
		a := &model.Attempt{OutTradeNo: no, Channel: "alipay", AccountID: acct, ProviderRef: "R-" + no,
			AmountMinor: minor, Currency: "CNY", Status: st, PaidAt: &paid}
		if err := s.CreateAttempt(a); err != nil {
			t.Fatalf("attempt %s: %v", no, err)
		}
	}
	mk("A", "acct-1", 10000, model.AttemptPaid, 1*time.Hour)   // 今日,计入
	mk("B", "acct-1", 5000, model.AttemptPaid, 2*time.Hour)    // 今日,计入 → acct-1=15000
	mk("C", "acct-2", 7000, model.AttemptPaid, 30*time.Minute) // 今日 acct-2=7000
	mk("D", "acct-1", 9999, model.AttemptPending, 10*time.Minute) // 未付,不计
	mk("E", "acct-1", 8888, model.AttemptPaid, 20*time.Hour)   // 昨天(paid_at < dayStart),不计

	got, err := s.SumPaidAttemptMinorByAccountSince(dayStart)
	if err != nil {
		t.Fatalf("sum: %v", err)
	}
	if got["acct-1"] != 15000 || got["acct-2"] != 7000 {
		t.Fatalf("聚合 = %+v, want acct-1=15000 acct-2=7000", got)
	}
}

internal/reconcile/usage_test.go:

package reconcile_test

import (
	"context"
	"testing"
	"time"

	"github.com/wangjia/pay/internal/accounts"
	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/reconcile"
	"github.com/wangjia/pay/internal/store"
)

func TestUsageSourceRefreshAndInterface(t *testing.T) {
	db := model.OpenTestDB(t)
	s := store.NewOrderStore(db)
	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	paid := now.Add(-time.Hour)
	_ = s.CreateAttempt(&model.Attempt{OutTradeNo: "A", Channel: "alipay", AccountID: "acct-1",
		ProviderRef: "R-A", AmountMinor: 12000, Currency: "CNY", Status: model.AttemptPaid, PaidAt: &paid})

	u := reconcile.NewUsageSource(s, func() time.Time { return now })
	var _ accounts.UsageSource = u // 编译期断言满足接口

	if u.TodayUsedMinor("acct-1") != 0 {
		t.Fatalf("刷新前应 0")
	}
	if err := u.Refresh(context.Background()); err != nil {
		t.Fatalf("refresh: %v", err)
	}
	if u.TodayUsedMinor("acct-1") != 12000 {
		t.Fatalf("刷新后 acct-1 应 12000, got %d", u.TodayUsedMinor("acct-1"))
	}
	if u.TodayUsedMinor("unknown") != 0 {
		t.Fatalf("未知账户应 0")
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/reconcile/ 2>&1 | tail -20 Expected: 编译失败——SumPaidAttemptMinorByAccountSince / reconcile.NewUsageSource 未定义。

• Step 3: 写实现

internal/store/order_query.go 追加:

// SumPaidAttemptMinorByAccountSince 聚合各账户自 since 起的已付金额(minor),供 LimitAware
// 判当日用量。量纲:attempt.AmountMinor 即账户所属渠道结算币种 minor(与 DailyLimit 同量纲)。
func (s *OrderStore) SumPaidAttemptMinorByAccountSince(since time.Time) (map[string]int64, error) {
	type row struct {
		AccountID string
		Total     int64
	}
	var rows []row
	if err := s.db.Model(&model.Attempt{}).
		Select("account_id, SUM(amount_minor) AS total").
		Where("status = ? AND account_id <> '' AND paid_at >= ?", model.AttemptPaid, since).
		Group("account_id").Scan(&rows).Error; err != nil {
		return nil, fmt.Errorf("store.SumPaidAttemptMinorByAccountSince: %w", err)
	}
	out := make(map[string]int64, len(rows))
	for _, r := range rows {
		out[r.AccountID] = r.Total
	}
	return out, nil
}

internal/reconcile/usage.go:

package reconcile

import (
	"context"
	"sync"
	"time"

	"github.com/wangjia/pay/internal/store"
)

// UsageSource 满足 accounts.UsageSource:持「账户→当日已收 minor」快照,对账 job 周期 Refresh。
// Pick 路径读快照(零 DB);Refresh 从 attempts 聚合。跨天由 now() 的自然日窗自动滚动。
type UsageSource struct {
	orders *store.OrderStore
	now    func() time.Time
	mu     sync.RWMutex
	snap   map[string]int64
}

func NewUsageSource(orders *store.OrderStore, now func() time.Time) *UsageSource {
	if now == nil {
		now = time.Now
	}
	return &UsageSource{orders: orders, now: now, snap: map[string]int64{}}
}

// TodayUsedMinor 读快照(accounts.UsageSource 接口);未刷新/未知账户返回 0(不误拒)。
func (u *UsageSource) TodayUsedMinor(accountID string) int64 {
	u.mu.RLock()
	defer u.mu.RUnlock()
	return u.snap[accountID]
}

// Refresh 从 attempts 重算当日快照(幂等覆盖)。当日起点用 now() 的 UTC 日期。
func (u *UsageSource) Refresh(ctx context.Context) error {
	m, err := u.orders.SumPaidAttemptMinorByAccountSince(StartOfDay(u.now()))
	if err != nil {
		return err
	}
	u.mu.Lock()
	u.snap = m
	u.mu.Unlock()
	return nil
}

// StartOfDay 返回 t 所在 UTC 自然日 00:00(DailyLimit 按自然日结算)。
func StartOfDay(t time.Time) time.Time {
	y, mo, d := t.UTC().Date()
	return time.Date(y, mo, d, 0, 0, 0, 0, time.UTC)
}

// RefreshUsageTask 把 Refresh 包成周期任务体。
func RefreshUsageTask(u *UsageSource) func(ctx context.Context) error {
	return func(ctx context.Context) error { return u.Refresh(ctx) }
}

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go build ./... && go test ./internal/store/ ./internal/reconcile/ ./internal/accounts/ -v 2>&1 | tail -30 Expected: 全 PASS(含 accounts 既有 limit_aware 测试仍绿)。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/store/order_query.go internal/store/order_query_test.go internal/reconcile/usage.go internal/reconcile/usage_test.go
git commit -m "feat(v2): LimitAware 真实用量源——DBUsageSource 快照 + 对账 job 周期聚合当日已收(替 NopUsage)"

---

Task 5: 对账主体 —— 周期 SyncPendingAttempts + 已付订单抽查 + main 装配 Runner

债务 #6(+ 继承 #7 已做的 settle-sync 日志)。 对账主体两条腿:① 逐 pending attempt 查单收敛——直接调度 P2 已有的 gateway.SyncPendingAttempts(内部已 query 每笔 pending、settle 命中、并对 not_found/mismatch/failed 打日志 = 债务 #7,764ed55);② 已付订单抽查核对——对近期已付的 attempt 反向 Query 渠道,比对金额/币种,发现漂移(如渠道侧已退款/拒付但本地仍 paid)即记日志告警。决策:全量渠道对账单/流水列表下载(alipay bill.download/stripe BalanceTransactions)判定过重(拉大文件、解析对账单格式、跨日切分),本期降级为「逐 pending attempt query + 已付订单抽查」,已覆盖「防掉单(query 收敛)+ 防重复授权/漂移(抽查)」两大风险,论证见 Self-Review。最后把 Task 1–4 与本任务的 job 全部挂上 reconcile.Runner,接进 main.go。

Files:

• Modify: internal/store/order_query.go(ListRecentlyPaidAttempts)
• Modify: internal/store/order_query_test.go
• Create: internal/reconcile/sync.go(SyncPendingTask + PaidSpotCheckTask)
• Create: internal/reconcile/sync_test.go
• Modify: config/config.go(ReconcileConfig + 默认值)
• Modify: main.go(装配 Runner + 各 job + crypto Warm + DBUsageSource 注入 Router)

Interfaces:

• store.(*OrderStore).ListRecentlyPaidAttempts(since time.Time, limit int) ([]model.Attempt, error)(status=paid AND paid_at>=since)。
• reconcile.SyncPendingTask(gw *gateway.Gateway, limit int) func(ctx) error。
• reconcile.PaidSpotCheckTask(orders *store.OrderStore, providers *provider.Registry, window time.Duration, now func() time.Time) func(ctx) error。
• config.ReconcileConfig{Enabled bool; OrderTTLMin, ExpireEverySec, SyncEverySec, UsageEverySec, SpotCheckEverySec, SpotCheckWindowMin int}。

抽查只日志告警不改状态:退款/拒付的状态机翻转属 P4,本期仅发现并可见。

• Step 1: 写失败测试

internal/store/order_query_test.go 追加:

func TestListRecentlyPaidAttempts(t *testing.T) {
	db := model.OpenTestDB(t)
	s := NewOrderStore(db)
	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	mk := func(no string, st model.AttemptStatus, paidAgo time.Duration) {
		paid := now.Add(-paidAgo)
		_ = s.CreateAttempt(&model.Attempt{OutTradeNo: no, Channel: "fake", ProviderRef: "R-" + no,
			AmountMinor: 100, Currency: "USDT", Status: st, PaidAt: &paid})
	}
	mk("RECENT", model.AttemptPaid, 10*time.Minute) // 近期已付 → 命中
	mk("OLD", model.AttemptPaid, 5*time.Hour)       // 太旧 → 不命中
	mk("PEND", model.AttemptPending, 1*time.Minute) // 未付 → 不命中

	got, err := s.ListRecentlyPaidAttempts(now.Add(-time.Hour), 50)
	if err != nil {
		t.Fatalf("list: %v", err)
	}
	if len(got) != 1 || got[0].OutTradeNo != "RECENT" {
		t.Fatalf("只应含 RECENT, got %+v", got)
	}
}

internal/reconcile/sync_test.go(用 fake provider + gateway,复用其装配式;这里给独立最小装配):

package reconcile_test

import (
	"context"
	"testing"
	"time"

	"github.com/wangjia/pay/config"
	"github.com/wangjia/pay/internal/accounts"
	"github.com/wangjia/pay/internal/gateway"
	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/provider"
	"github.com/wangjia/pay/internal/provider/fake"
	"github.com/wangjia/pay/internal/reconcile"
	"github.com/wangjia/pay/internal/store"
)

type stubResolver struct{}

func (stubResolver) Resolve(sku, currency string) (int64, string, string, error) {
	return 29990000, "Pro", "pro_year", nil
}

type nopEnq struct{}

func (nopEnq) Enqueue(_, _, _ string, _ map[string]any) error { return nil }

func TestSyncPendingTaskSettlesViaQuery(t *testing.T) {
	db := model.OpenTestDB(t)
	orders := store.NewOrderStore(db)
	preg := provider.NewRegistry()
	fp := fake.New()
	preg.Register(fp)
	areg := accounts.New([]config.AccountConfig{{AccountID: "fake-a1", Channel: "fake", Region: "global", Enabled: true}})
	gw := gateway.New(orders, preg, accounts.NewRouter(areg, nil, nil), stubResolver{}, nopEnq{}, "global")

	res, _ := gw.CreateOrder(context.Background(), gateway.CreateOrderInput{SKU: "pro_year", Method: "fake", BizSystem: "pangolin", BizRef: "u-1"})
	atts, _ := orders.ListAttemptsByStatus(model.AttemptPending, 10)
	fp.SetQueryResult(atts[0].ProviderRef, provider.PaidEvent{
		ProviderRef: atts[0].ProviderRef, Status: provider.PaidSucceeded, PaidAmountMinor: 29990000, PaidCurrency: "USDT"})

	task := reconcile.SyncPendingTask(gw, 50)
	if err := task(context.Background()); err != nil {
		t.Fatalf("task: %v", err)
	}
	o, _ := orders.GetOrder(res.OrderNo)
	if o.Status != model.OrderPaidV2 {
		t.Fatalf("查单对账后应 paid, got %v", o.Status)
	}
	_ = time.Second
}

func TestPaidSpotCheckTaskRunsCleanOnConsistent(t *testing.T) {
	db := model.OpenTestDB(t)
	orders := store.NewOrderStore(db)
	preg := provider.NewRegistry()
	fp := fake.New()
	preg.Register(fp)
	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	paid := now.Add(-10 * time.Minute)
	_ = orders.CreateAttempt(&model.Attempt{OutTradeNo: "O1", Channel: "fake", ProviderRef: "R-O1",
		AmountMinor: 100, Currency: "USDT", Status: model.AttemptPaid, PaidAt: &paid})
	// 渠道侧查单仍报 succeeded 同额 → 一致,无告警。
	fp.SetQueryResult("R-O1", provider.PaidEvent{ProviderRef: "R-O1", Status: provider.PaidSucceeded, PaidAmountMinor: 100, PaidCurrency: "USDT"})

	task := reconcile.PaidSpotCheckTask(orders, preg, time.Hour, func() time.Time { return now })
	if err := task(context.Background()); err != nil {
		t.Fatalf("spotcheck: %v", err) // 只求不报错、不 panic;漂移检测走日志
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/reconcile/ 2>&1 | tail -20 Expected: 编译失败——ListRecentlyPaidAttempts / SyncPendingTask / PaidSpotCheckTask 未定义。

• Step 3: 写实现

internal/store/order_query.go 追加:

// ListRecentlyPaidAttempts 列近期(paid_at>=since)已付 attempt,供对账抽查反查渠道核对。
func (s *OrderStore) ListRecentlyPaidAttempts(since time.Time, limit int) ([]model.Attempt, error) {
	if limit <= 0 || limit > 500 {
		limit = 100
	}
	var out []model.Attempt
	if err := s.db.Where("status = ? AND paid_at >= ?", model.AttemptPaid, since).
		Order("id DESC").Limit(limit).Find(&out).Error; err != nil {
		return nil, fmt.Errorf("store.ListRecentlyPaidAttempts: %w", err)
	}
	return out, nil
}

internal/reconcile/sync.go:

package reconcile

import (
	"context"
	"log"
	"time"

	"github.com/wangjia/pay/internal/gateway"
	"github.com/wangjia/pay/internal/provider"
	"github.com/wangjia/pay/internal/store"
)

// SyncPendingTask 调度 P2 gateway.SyncPendingAttempts:逐 pending attempt 查单收敛(防掉单)。
// 其内部已对 not_found/amount_mismatch/failed 打日志(settle-sync,764ed55),此处不重复。
func SyncPendingTask(gw *gateway.Gateway, limit int) func(ctx context.Context) error {
	return func(ctx context.Context) error {
		n, err := gw.SyncPendingAttempts(ctx, limit)
		if err != nil {
			return err
		}
		if n > 0 {
			log.Printf("[reconcile] 查单对账收敛 %d 笔待支付 → paid", n)
		}
		return nil
	}
}

// PaidSpotCheckTask 已付订单抽查:对近 window 内已付 attempt 反查渠道,金额/币种漂移即告警
// (如渠道侧已退款/拒付而本地仍 paid)。只发现不改状态——状态机翻转属 P4。
// crypto 之类 query-only 渠道:paid 后再查若命中同额即一致;查不到(链上历史滚出窗口)不报错跳过。
func PaidSpotCheckTask(orders *store.OrderStore, providers *provider.Registry, window time.Duration, now func() time.Time) func(ctx context.Context) error {
	return func(ctx context.Context) error {
		atts, err := orders.ListRecentlyPaidAttempts(now().Add(-window), 100)
		if err != nil {
			return err
		}
		for i := range atts {
			a := &atts[i]
			prov, err := providers.Get(a.Channel)
			if err != nil {
				continue
			}
			created := a.CreatedAt
			ev, err := prov.Query(ctx, provider.QueryRequest{
				ProviderRef: a.ProviderRef, OutTradeNo: a.OutTradeNo, AccountID: a.AccountID,
				AmountMinor: a.AmountMinor, Currency: a.Currency, CreatedAt: created, ExpiresAt: a.ExpiresAt,
			})
			if err != nil || ev == nil {
				continue // 查不到/瞬时错:抽查尽力而为,不阻断
			}
			// 本地 paid,渠道却报非成功,或金额/币种对不上 → 对账差异,必须可见。
			if ev.Status != provider.PaidSucceeded || ev.PaidCurrency != a.Currency || ev.PaidAmountMinor < a.AmountMinor {
				log.Printf("[reconcile][对账差异] attempt=%s channel=%s 本地 paid 但渠道 status=%s amount=%d/%s(本地 %d/%s)",
					a.ProviderRef, a.Channel, ev.Status, ev.PaidAmountMinor, ev.PaidCurrency, a.AmountMinor, a.Currency)
			}
		}
		return nil
	}
}

config/config.go —— Config 加字段 + 结构体 + 默认值:

// Config 结构体加一行:
	QuerySync     QuerySyncConfig     `mapstructure:"query_sync"`
	Reconcile     ReconcileConfig     `mapstructure:"reconcile"`

// ReconcileConfig v2 后台守护/对账周期任务开关与间隔。
type ReconcileConfig struct {
	Enabled            bool `mapstructure:"enabled"`
	OrderTTLMin        int  `mapstructure:"order_ttl_min"`         // pending 订单存活 TTL(分钟),超则关闭
	ExpireEverySec     int  `mapstructure:"expire_every_sec"`      // 过期清理间隔
	SyncEverySec       int  `mapstructure:"sync_every_sec"`        // 查单对账间隔
	UsageEverySec      int  `mapstructure:"usage_every_sec"`       // 用量快照刷新间隔
	SpotCheckEverySec  int  `mapstructure:"spot_check_every_sec"`  // 已付抽查间隔
	SpotCheckWindowMin int  `mapstructure:"spot_check_window_min"` // 抽查回溯窗(分钟)
	OrphanEverySec     int  `mapstructure:"orphan_every_sec"`      // crypto 孤儿扫描间隔(Task 6)
	OrphanWindowMin    int  `mapstructure:"orphan_window_min"`     // 孤儿扫描回溯窗(Task 6)
}

Load() 里补默认(在 query_sync 默认附近):

	viper.SetDefault("reconcile.enabled", true)
	viper.SetDefault("reconcile.order_ttl_min", 60)
	viper.SetDefault("reconcile.expire_every_sec", 300)
	viper.SetDefault("reconcile.sync_every_sec", 30)
	viper.SetDefault("reconcile.usage_every_sec", 60)
	viper.SetDefault("reconcile.spot_check_every_sec", 300)
	viper.SetDefault("reconcile.spot_check_window_min", 180)
	viper.SetDefault("reconcile.orphan_every_sec", 300)
	viper.SetDefault("reconcile.orphan_window_min", 180)

main.go —— 装配(替换现有 notifier/Router/gateway 段并追加 Runner)。关键改动:

1. import "context" + "github.com/wangjia/pay/internal/reconcile" + "github.com/wangjia/pay/internal/provider/crypto"(crypto 仅用于 Warm 类型断言,见下 Task 6 亦需);
2. 用 reconcile.NewUsageSource 替 accounts.NopUsage{} 注入 Router;
3. crypto Warm(若注册了 crypto):给 providerbuild.BuildRegistry 传 loader,或建后取 crypto 实例 Warm。装配式:crypto 的 loader 需要 orderStore,故 crypto Provider 的构造要能带 WithReservationLoader。若 providerbuild.BuildRegistry 不便传 loader,改为 Warm 前对注册表里的 crypto 实例调 Warm——为此让 crypto 在 build 时就注入 loader。最小改动:在 providerbuild 暴露把 loader 传进 crypto.New 的入口,或 main 直接从 pReg.Get("crypto") 拿到 *crypto.Provider 调 Warm(loader 用 reconcile.CryptoReservationLoader(orderStore),但 loader 是构造期注入的……)。决策:给 crypto.(*Provider) 的 loader 支持构造后注入——WithReservationLoader 已是 Option 构造期注入;为让 main 在 build 之后注入 loader,补一个 setter SetReservationLoader(l)。见下 Step 3a。

	orderStore := store.NewOrderStore(db)
	webhookStore := store.NewWebhookStore(db)
	notifier := webhook.NewNotifier(webhookStore, config.C.BizByName, func(no string) (bool, error) {
		o, err := orderStore.GetOrder(no)
		if err != nil {
			return false, err
		}
		return o.Status == model.OrderPaidV2, nil
	}) // 硬化选项走默认(退避/死信/告警);如需覆盖用 webhook.WithMaxAttempts 等
	notifier.Start(60 * time.Second)

	productResolver := gateway.NewDBProductResolver(db)
	acctReg := accounts.New(config.C.Accounts)
	pReg := providerbuild.BuildRegistry(acctReg)
	usage := reconcile.NewUsageSource(orderStore, time.Now) // P6 真实用量源替 NopUsage
	acctPicker := accounts.NewRouter(acctReg, config.C.Routing, usage)
	gw := gateway.New(orderStore, pReg, acctPicker, productResolver, notifier, "cn")
	router.SetupV2(r, gw)

	// P6 后台守护 / 对账
	if config.C.Reconcile.Enabled {
		rc := config.C.Reconcile
		runner := reconcile.NewRunner()
		runner.Add("order-expire", time.Duration(rc.ExpireEverySec)*time.Second,
			reconcile.OrderExpirerTask(orderStore, time.Duration(rc.OrderTTLMin)*time.Minute, time.Now))
		runner.Add("usage-refresh", time.Duration(rc.UsageEverySec)*time.Second,
			reconcile.RefreshUsageTask(usage))
		runner.Add("sync-pending", time.Duration(rc.SyncEverySec)*time.Second,
			reconcile.SyncPendingTask(gw, 100))
		runner.Add("paid-spotcheck", time.Duration(rc.SpotCheckEverySec)*time.Second,
			reconcile.PaidSpotCheckTask(orderStore, pReg, time.Duration(rc.SpotCheckWindowMin)*time.Minute, time.Now))
		// crypto 孤儿扫描(Task 6)在此追加。
		reconcile.AddCryptoJobs(runner, pReg, acctReg, orderStore, rc) // Task 6 提供;含 Warm + 孤儿扫描注册

		ctx := context.Background()
		runner.RunOnce(ctx) // 启动预热:先跑一遍(usage 快照/过期清理立即生效)
		runner.Start(ctx)
		log.Printf("[reconcile] 后台守护已启动(过期清理/用量刷新/查单对账/已付抽查/孤儿扫描)")
	}

reconcile.AddCryptoJobs 与 crypto Warm/孤儿逻辑落在 Task 6;本任务先让 main 编译通过可临时用一个占位……不:禁占位。改为 Task 5 只装配前四个 job + runner.RunOnce/Start,把 AddCryptoJobs 这一行留到 Task 6 再加(Task 6 Step 3 明确「在 main 追加此行」)。故本 Step 3 的 main 片段删掉 AddCryptoJobs 那行,Task 6 再插回。

• Step 3a: crypto loader 构造后注入(供 main Warm 用)

internal/provider/crypto/crypto.go 补 setter(Task 6 的 Warm 装配依赖):

// SetReservationLoader 构造后注入冷启动预留源(装配期 main 在 BuildRegistry 之后调用:
// loader 依赖 OrderStore,而注册表构造不便传 store)。非并发安全,仅启动期单线程调用。
func (p *Provider) SetReservationLoader(l ReservationLoader) { p.loader = l }

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go build ./... && go test ./internal/store/ ./internal/reconcile/ ./internal/gateway/ -v 2>&1 | tail -30 Expected: 全 PASS;go build ./...(main 装配)通过。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/store/order_query.go internal/store/order_query_test.go internal/reconcile/sync.go internal/reconcile/sync_test.go internal/provider/crypto/crypto.go config/config.go main.go
git commit -m "feat(v2): 对账主体——周期查单收敛 + 已付订单抽查 + main 装配 reconcile Runner(前 4 job)"

---

Task 6: crypto 孤儿链上支付发现(OrphanScanner + orphan 记录/告警)

债务 #3。 到账但不匹配任何 attempt 的转账(付款人转错金额、手动转账、迟到到一笔金额早已滚出所有活跃单的旧款)必须被发现并记录,不能静默。canonical(pangolin-pay watcher)把这类记 orphan_payments 供人工对账。本任务:中性可选接口 provider.OrphanScanner,crypto 实现(扫地址近 window 确认到账,金额不在「近期任一 attempt 的期望金额集」内 → orphan);reconcile 收集 crypto 账户的已知期望金额、调 ScanOrphans、把 orphan 落 model.OrphanPayment(tx_id unique 去重)+ 告警。同时把 crypto Warm 也在此接进 main(AddCryptoJobs)。

Files:

• Modify: internal/provider/provider.go(中性 OrphanScanner 接口 + OrphanScanRequest/OrphanTransfer/KnownAttempt)
• Modify: internal/provider/crypto/crypto.go(实现 ScanOrphans)
• Modify: internal/provider/crypto/crypto_test.go(httptest 假 TronGrid:一笔匹配 + 一笔孤儿)
• Create: internal/model/orphan.go(OrphanPayment 表)
• Modify: main.go(AutoMigrate 加表)
• Create: internal/store/orphan.go(OrphanStore.Record 幂等)
• Create: internal/store/orphan_test.go
• Create: internal/reconcile/orphan.go(AddCryptoJobs:Warm + OrphanScanTask)
• Create: internal/reconcile/orphan_test.go
• Modify: main.go(追加 reconcile.AddCryptoJobs(...) 那行 + import crypto/context 已在 Task 5)

Interfaces:

• provider.OrphanScanRequest{AccountID string; Since time.Time; Known []KnownAttempt};provider.KnownAttempt{AmountMinor int64; ProviderRef string}(期望金额 = base + tail,tail 由渠道自解 ref,pay 不算);provider.OrphanTransfer{TxID string; AmountMinor int64; Currency string; At time.Time};provider.OrphanScanner interface { ScanOrphans(ctx, OrphanScanRequest) ([]OrphanTransfer, error) }。

• store.(*OrderStore) 复用 ListAttemptsByChannelSince(新增:任意状态、channel、created_at>=since)——arphan 的「已知期望金额集」含 pending/paid/expired 全部,凡 pay 合法签发过的金额都不算孤儿。

• store.(*OrphanStore).Record(o *model.OrphanPayment) (bool, error)(tx_id 冲突 no-op,返回是否新记)。

• reconcile.AddCryptoJobs(runner *Runner, providers *provider.Registry, accts *accounts.Registry, orders *store.OrderStore, cfg config.ReconcileConfig)。

• Step 1: 写失败测试

internal/store/order_query.go 先补 ListAttemptsByChannelSince(Task 6 需要),测试放 order_query_test.go:

func TestListAttemptsByChannelSince(t *testing.T) {
	db := model.OpenTestDB(t)
	s := NewOrderStore(db)
	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	mk := func(no, ch string, ago time.Duration) {
		_ = s.CreateAttempt(&model.Attempt{OutTradeNo: no, Channel: ch, ProviderRef: "R-" + no,
			AmountMinor: 100, Currency: "USDT", Status: model.AttemptPending})
		_ = db.Model(&model.Attempt{}).Where("out_trade_no = ?", no).Update("created_at", now.Add(-ago)).Error
	}
	mk("C1", "crypto", 10*time.Minute)
	mk("C2", "crypto", 5*time.Hour) // 太旧
	mk("A1", "alipay", 1*time.Minute)

	got, err := s.ListAttemptsByChannelSince("crypto", now.Add(-time.Hour), 100)
	if err != nil {
		t.Fatalf("list: %v", err)
	}
	if len(got) != 1 || got[0].OutTradeNo != "C1" {
		t.Fatalf("只应含近期 crypto, got %+v", got)
	}
}

internal/store/orphan_test.go:

package store_test

import (
	"testing"
	"time"

	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/store"
)

func TestOrphanStoreRecordIdempotent(t *testing.T) {
	os := store.NewOrphanStore(model.OpenTestDB(t))
	o := &model.OrphanPayment{Channel: "crypto", AccountID: "cry-1", TxID: "TX-1",
		AmountMinor: 12345, Currency: "USDT", DetectedAt: time.Now()}
	first, err := os.Record(o)
	if err != nil || !first {
		t.Fatalf("首次应记入, first=%v err=%v", first, err)
	}
	again, err := os.Record(&model.OrphanPayment{Channel: "crypto", AccountID: "cry-1", TxID: "TX-1",
		AmountMinor: 12345, Currency: "USDT", DetectedAt: time.Now()})
	if err != nil || again {
		t.Fatalf("同 tx_id 应幂等 no-op, again=%v err=%v", again, err)
	}
}

internal/provider/crypto/crypto_test.go 追加(假 TronGrid 返回两笔:一笔金额匹配 known、一笔不匹配 → 仅后者为 orphan):

func TestScanOrphansFlagsUnmatchedTransfer(t *testing.T) {
	const addr = "TOrphanScanAddr00000000000000000000"
	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	// 假 TronGrid:to=addr 两笔确认到账。29990263 匹配 known(base 29990000 + tail 263);88880000 无主。
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		_, _ = w.Write([]byte(`{"data":[
			{"transaction_id":"TX-MATCH","to":"` + addr + `","type":"Transfer","value":"29990263","block_timestamp":` + strconv.FormatInt(now.Add(-5*time.Minute).UnixMilli(), 10) + `},
			{"transaction_id":"TX-ORPHAN","to":"` + addr + `","type":"Transfer","value":"88880000","block_timestamp":` + strconv.FormatInt(now.Add(-3*time.Minute).UnixMilli(), 10) + `}
		]}`))
	}))
	defer ts.Close()

	t.Setenv("CRY_ADDRESS", addr)
	reg := accounts.New([]config.AccountConfig{{AccountID: "cry-1", Channel: "crypto", Enabled: true, CredentialEnvPrefix: "cry"}})
	p := crypto.New(reg, crypto.WithBaseURL(ts.URL), crypto.WithHTTPClient(ts.Client()), crypto.WithNow(func() time.Time { return now }))

	orphans, err := p.ScanOrphans(context.Background(), provider.OrphanScanRequest{
		AccountID: "cry-1", Since: now.Add(-time.Hour),
		Known: []provider.KnownAttempt{{AmountMinor: 29990000, ProviderRef: "CRYPTO-PAY-A-263"}}, // 期望 29990263
	})
	if err != nil {
		t.Fatalf("scan: %v", err)
	}
	if len(orphans) != 1 || orphans[0].TxID != "TX-ORPHAN" || orphans[0].AmountMinor != 88880000 {
		t.Fatalf("只应报 1 笔孤儿 TX-ORPHAN, got %+v", orphans)
	}
}

internal/reconcile/orphan_test.go(loader→scan→record 全链;用上面的假 TronGrid 同构):

package reconcile_test

import (
	"context"
	"net/http"
	"net/http/httptest"
	"strconv"
	"testing"
	"time"

	"github.com/wangjia/pay/config"
	"github.com/wangjia/pay/internal/accounts"
	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/provider"
	"github.com/wangjia/pay/internal/provider/crypto"
	"github.com/wangjia/pay/internal/reconcile"
	"github.com/wangjia/pay/internal/store"
)

func TestOrphanScanTaskRecordsUnmatched(t *testing.T) {
	const addr = "TOrphanJobAddr0000000000000000000000"
	now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC)
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte(`{"data":[
			{"transaction_id":"TX-ORPHAN","to":"` + addr + `","type":"Transfer","value":"88880000","block_timestamp":` + strconv.FormatInt(now.Add(-3*time.Minute).UnixMilli(), 10) + `}
		]}`))
	}))
	defer ts.Close()

	t.Setenv("CRY_ADDRESS", addr)
	db := model.OpenTestDB(t)
	orders := store.NewOrderStore(db)
	orphanStore := store.NewOrphanStore(db)
	acctReg := accounts.New([]config.AccountConfig{{AccountID: "cry-1", Channel: "crypto", Enabled: true, CredentialEnvPrefix: "cry"}})
	preg := provider.NewRegistry()
	preg.Register(crypto.New(acctReg, crypto.WithBaseURL(ts.URL), crypto.WithHTTPClient(ts.Client()), crypto.WithNow(func() time.Time { return now })))

	task := reconcile.OrphanScanTask(preg, acctReg, orders, orphanStore, time.Hour, func() time.Time { return now })
	if err := task(context.Background()); err != nil {
		t.Fatalf("task: %v", err)
	}
	var cnt int64
	db.Model(&model.OrphanPayment{}).Where("tx_id = ?", "TX-ORPHAN").Count(&cnt)
	if cnt != 1 {
		t.Fatalf("应落 1 条孤儿, got %d", cnt)
	}
}

• Step 2: 跑测试确认失败

Run: cd /Users/wangjia/code/pay && go test ./internal/store/ ./internal/provider/crypto/ ./internal/reconcile/ 2>&1 | tail -20 Expected: 编译失败——OrphanScanner/OrphanScanRequest/OrphanPayment/OrphanStore/OrphanScanTask/ListAttemptsByChannelSince 未定义。

• Step 3: 写实现

internal/provider/provider.go 追加(中性类型 + 可选接口,类比 RefundingProvider):

// ---- 对账:孤儿到账扫描(P6,可选接口)----

// KnownAttempt 是 pay 合法签发过的一笔尝试的对账维度:期望金额 = base(AmountMinor)+ 渠道尾数
// (尾数封在 provider_ref,由渠道自解,pay 不算)。渠道据此判断一笔到账是否"有主"。
type KnownAttempt struct {
	AmountMinor int64
	ProviderRef string
}

// OrphanScanRequest 扫描某账户 Since 以来、不匹配任何 Known 的到账。
type OrphanScanRequest struct {
	AccountID string
	Since     time.Time
	Known     []KnownAttempt
}

// OrphanTransfer 一笔"有钱到账但无主"的转账(付错金额/手动转/超窗迟到旧款)。
type OrphanTransfer struct {
	TxID        string
	AmountMinor int64
	Currency    string
	At          time.Time
}

// OrphanScanner 自托管渠道(crypto)可选实现:发现到账但不匹配任何 attempt 的转账。
// 网关侧渠道(alipay/stripe)以对账单核对,不实现此接口。
type OrphanScanner interface {
	ScanOrphans(ctx context.Context, req OrphanScanRequest) ([]OrphanTransfer, error)
}

internal/provider/crypto/crypto.go 实现 ScanOrphans(复用 Query 里的 TronGrid 请求构造 + tailFromRef):

// ScanOrphans 扫地址近 Since 的确认到账,金额不在"任一 Known 的期望金额集"内 → 孤儿。
// 期望金额 = known.AmountMinor + tailFromRef(known.ProviderRef);块时须晚于 Since。
func (p *Provider) ScanOrphans(ctx context.Context, req provider.OrphanScanRequest) ([]provider.OrphanTransfer, error) {
	addr, err := p.address(req.AccountID)
	if err != nil {
		return nil, err
	}
	expected := make(map[int64]struct{}, len(req.Known))
	for _, k := range req.Known {
		tail, terr := tailFromRef(k.ProviderRef)
		if terr != nil {
			continue // 无尾数的 ref 跳过(不误判为孤儿依据)
		}
		expected[k.AmountMinor+tail] = struct{}{}
	}

	endpoint := fmt.Sprintf("%s/v1/accounts/%s/transactions/trc20?only_confirmed=true&contract_address=%s&limit=50",
		p.baseURL, url.PathEscape(addr), url.QueryEscape(USDTContract))
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
	if err != nil {
		return nil, err
	}
	if k := p.apiKey(req.AccountID); k != "" {
		httpReq.Header.Set("TRON-PRO-API-KEY", k)
	}
	resp, err := p.http.Do(httpReq)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("crypto: TronGrid HTTP %d: %s", resp.StatusCode, body)
	}
	var tr trc20Resp
	if err := json.Unmarshal(body, &tr); err != nil {
		return nil, err
	}

	sinceUnix := req.Since.Unix()
	var out []provider.OrphanTransfer
	for _, d := range tr.Data {
		if d.To != addr || d.Type != "Transfer" {
			continue
		}
		blockTs := d.BlockMs / 1000
		if blockTs < sinceUnix {
			continue // 窗外旧款不扫(避免把历史正常单反复报孤儿)
		}
		val, perr := strconv.ParseInt(d.Value, 10, 64)
		if perr != nil {
			continue
		}
		if _, ok := expected[val]; ok {
			continue // 金额有主(匹配某 attempt 期望额)→ 非孤儿
		}
		out = append(out, provider.OrphanTransfer{
			TxID: d.TxID, AmountMinor: val, Currency: "USDT", At: time.Unix(blockTs, 0),
		})
	}
	return out, nil
}

internal/model/orphan.go:

package model

import "time"

// OrphanPayment 到账但不匹配任何 attempt 的转账(对账发现,供人工核对/退款)。
// tx_id 唯一 → 重复扫描幂等 no-op。canonical pangolin-pay orphan_payments 的 v2 对应物。
type OrphanPayment struct {
	Base
	Channel     string    `gorm:"index;size:32;not null"`
	AccountID   string    `gorm:"index;size:64"`
	TxID        string    `gorm:"uniqueIndex;size:128;not null"`
	AmountMinor int64     `gorm:"not null"`
	Currency    string    `gorm:"size:16;not null"`
	DetectedAt  time.Time
	Note        string `gorm:"size:255"`
}

main.go 的 autoMigrate 参数表加 &model.OrphanPayment{}。

internal/store/orphan.go:

package store

import (
	"fmt"

	"gorm.io/gorm"
	"gorm.io/gorm/clause"

	"github.com/wangjia/pay/internal/model"
)

type OrphanStore struct{ db *gorm.DB }

func NewOrphanStore(db *gorm.DB) *OrphanStore { return &OrphanStore{db: db} }

// Record 幂等落一条孤儿(tx_id 冲突 no-op)。返回是否新记(供告警只喊一次)。
func (s *OrphanStore) Record(o *model.OrphanPayment) (bool, error) {
	res := s.db.Clauses(clause.OnConflict{
		Columns:   []clause.Column{{Name: "tx_id"}},
		DoNothing: true,
	}).Create(o)
	if res.Error != nil {
		return false, fmt.Errorf("store.OrphanStore.Record: %w", res.Error)
	}
	return res.RowsAffected > 0, nil
}

internal/store/order_query.go 补 ListAttemptsByChannelSince:

// ListAttemptsByChannelSince 列某渠道 created_at>=since 的 attempt(任意状态),
// 供 orphan 扫描构造"已知期望金额集"(凡 pay 合法签发过的金额都不算孤儿)。
func (s *OrderStore) ListAttemptsByChannelSince(channel string, since time.Time, limit int) ([]model.Attempt, error) {
	if limit <= 0 || limit > 500 {
		limit = 200
	}
	var out []model.Attempt
	if err := s.db.Where("channel = ? AND created_at >= ?", channel, since).
		Order("id DESC").Limit(limit).Find(&out).Error; err != nil {
		return nil, fmt.Errorf("store.ListAttemptsByChannelSince: %w", err)
	}
	return out, nil
}

internal/reconcile/orphan.go:

package reconcile

import (
	"context"
	"log"
	"time"

	"github.com/wangjia/pay/config"
	"github.com/wangjia/pay/internal/accounts"
	"github.com/wangjia/pay/internal/model"
	"github.com/wangjia/pay/internal/provider"
	"github.com/wangjia/pay/internal/provider/crypto"
	"github.com/wangjia/pay/internal/store"
)

// OrphanScanTask 对每个 enabled crypto 账户扫链找孤儿到账(不匹配任何近期 attempt),
// 落 OrphanStore + 首次记入时告警。渠道须实现 provider.OrphanScanner(crypto 实现)。
func OrphanScanTask(providers *provider.Registry, accts *accounts.Registry, orders *store.OrderStore,
	orphans *store.OrphanStore, window time.Duration, now func() time.Time) func(ctx context.Context) error {
	return func(ctx context.Context) error {
		prov, err := providers.Get("crypto")
		if err != nil {
			return nil // 未启用 crypto:无事可做
		}
		scanner, ok := prov.(provider.OrphanScanner)
		if !ok {
			return nil
		}
		since := now().Add(-window)
		for _, acc := range accts.EnabledFor("crypto", "") {
			atts, err := orders.ListAttemptsByChannelSince("crypto", since, 200)
			if err != nil {
				return err
			}
			known := make([]provider.KnownAttempt, 0, len(atts))
			for i := range atts {
				if atts[i].AccountID != acc.AccountID {
					continue
				}
				known = append(known, provider.KnownAttempt{AmountMinor: atts[i].AmountMinor, ProviderRef: atts[i].ProviderRef})
			}
			found, err := scanner.ScanOrphans(ctx, provider.OrphanScanRequest{
				AccountID: acc.AccountID, Since: since, Known: known,
			})
			if err != nil {
				log.Printf("[reconcile] 孤儿扫描 account=%s: %v", acc.AccountID, err)
				continue // 单账户失败不阻断其它
			}
			for _, o := range found {
				isNew, rerr := orphans.Record(&model.OrphanPayment{
					Channel: "crypto", AccountID: acc.AccountID, TxID: o.TxID,
					AmountMinor: o.AmountMinor, Currency: o.Currency, DetectedAt: o.At,
					Note: "到账无主:不匹配任何近期 attempt 期望金额",
				})
				if rerr != nil {
					log.Printf("[reconcile] 记录孤儿失败 tx=%s: %v", o.TxID, rerr)
					continue
				}
				if isNew {
					log.Printf("[reconcile][孤儿告警] channel=crypto account=%s tx=%s amount=%d %s",
						acc.AccountID, o.TxID, o.AmountMinor, o.Currency)
				}
			}
		}
		return nil
	}
}

// AddCryptoJobs 装配 crypto 相关后台任务:① 冷启动 Warm(注入 loader 后立即重建预留);
// ② 周期孤儿扫描。若未启用 crypto 渠道则安全跳过。
func AddCryptoJobs(runner *Runner, providers *provider.Registry, accts *accounts.Registry,
	orders *store.OrderStore, orphans *store.OrphanStore, cfg config.ReconcileConfig) {
	prov, err := providers.Get("crypto")
	if err != nil {
		return
	}
	if cp, ok := prov.(*crypto.Provider); ok {
		cp.SetReservationLoader(CryptoReservationLoader(orders))
		if werr := cp.Warm(context.Background()); werr != nil { // 起服务前重建预留
			log.Printf("[reconcile] crypto 预留冷启动重建: %v", werr)
		}
	}
	runner.Add("crypto-orphan-scan", time.Duration(cfg.OrphanEverySec)*time.Second,
		OrphanScanTask(providers, accts, orders, orphans,
			time.Duration(cfg.OrphanWindowMin)*time.Minute, time.Now))
}

main.go —— 在 Task 5 装配段的注释处补一行(需 orphanStore := store.NewOrphanStore(db)):

		orphanStore := store.NewOrphanStore(db)
		reconcile.AddCryptoJobs(runner, pReg, acctReg, orderStore, orphanStore, rc)

（放在 runner.RunOnce(ctx) 之前,使 Warm 在起服务前完成、孤儿 job 一并进 runner。）

• Step 4: 跑测试确认通过

Run: cd /Users/wangjia/code/pay && go build ./... && go test ./... 2>&1 | tail -30 Expected: 全仓 PASS;go build ./... 通过。

• Step 5: Commit

cd /Users/wangjia/code/pay
git add internal/provider/provider.go internal/provider/crypto/crypto.go internal/provider/crypto/crypto_test.go internal/model/orphan.go internal/store/orphan.go internal/store/orphan_test.go internal/store/order_query.go internal/store/order_query_test.go internal/reconcile/orphan.go internal/reconcile/orphan_test.go main.go
git commit -m "feat(v2): crypto 孤儿到账发现——OrphanScanner 扫链核对 + orphan_payments 落表告警(对账兜底)"

---

Self-Review

Spec coverage(P6 范围,对照设计 §8「对账优先于实时」+ 三份前置计划 routed 到 P6 的债务):

• 订单级过期清理(债务 #1,Task 1)= 超 TTL pending 单条件 UPDATE 关闭,含零尝试孤儿单;与 attempt 级 ExpirePendingAttempts 语义严格区分 ✓。
• Notifier 硬化(债务 #2,Task 2)= 指数退避(base·2^(n-1) 封顶)+ 最大次数死信(dead 列)+ 告警钩子(WithAlerter,默认 log);投递门禁(未付不投不计失败)与 unique 键幂等不破 ✓。
• crypto 预留冷启动兜底(债务 #4,Task 3)= 注入 ReservationLoader 从 pending crypto attempts 重建 reserved,兜重启丢内存;crypto 不 import store(中性结构 + 装配层桥)✓。
• LimitAware 真实数据源(债务 #5,Task 4)= reconcile.UsageSource 满足 accounts.UsageSource,对账 job 周期从 attempts 聚合当日已收(按账户 = 渠道结算币种 minor,与 DailyLimit 同量纲);Router 注入它替 NopUsage,LimitAware/Router 零改动(兑现 P5 D5)✓。
• 对账主体(债务 #6,Task 5)= 周期 SyncPendingAttempts(查单收敛,防掉单;继承其 settle-sync 日志 = 债务 #7 已做,不重复)+ 已付订单抽查反查渠道(防重复授权/漂移)✓。
• crypto 孤儿发现(债务 #3,Task 6)= 中性 OrphanScanner,crypto 扫链找无主到账,落 OrphanPayment(tx_id unique 幂等)+ 首记告警 ✓。

关键决策:

• 后台 job 统一形态:reconcile.Runner 一任务一 ticker + recover + RunOnce(启动预热 + 单测入口);每任务幂等 sweep,崩溃后下 tick 自愈。时钟一律注入(now func() time.Time),cutoff/since/退避 Go 端算,SQL 不用时间函数(sqlite/mysql 可移植)。
• 对账深度(债务 #6 决策):全量渠道对账单下载判定过重——alipay bill.download(拉 CSV/ZIP、解析对账单格式、按结算日切分、处理手续费行)与 stripe BalanceTransactions 分页拉全量,工程量与运维面(存储/解析/对差)远超本期单实例小体量所需。降级为「逐 pending attempt query(已存在)+ 已付订单抽查反查」:前者防「掉单」(回调丢失→查单收敛),后者防「重复授权/状态漂移」(本地 paid 但渠道退款/拒付→日志告警)。crypto 因自托管无对账单,其「渠道流水核对」正是 Task 6 的扫链孤儿发现。这套已覆盖设计 §8 两大风险(防掉单 + 防重复授权),真需要全量对账单时再单开任务。
• orphan「已知金额集」取全状态近窗 attempt:pending/paid/expired 一律计入期望金额集——凡 pay 合法签发过的金额都不算孤儿(一笔匹配已 paid attempt 金额的到账 = 正常付款,非孤儿;匹配 pending/expired 的 = settle-sync 处理,非孤儿)。孤儿严格 = 「到账金额不在任何近窗 attempt 期望额内」(付错额/手动转/超窗旧款),精确对齐债务 #3 措辞。窗外旧款不扫,避免历史单反复误报。
• 抽查只发现不改状态:退款/拒付的订单态机翻转属 P4,本期仅日志告警使其可见(不越权改 P4 的状态语义)。
• Notifier options 向后兼容:NewNotifier(...opts) 变参,现有 3 参调用不改;ListUndelivered 保留(测试断言用),Notifier 内部改用退避感知的 ListDeliverable。

Provider 中性守恒: internal/gateway、internal/provider 核心不 import reconcile;crypto 不 import store。孤儿经中性 provider.OrphanScanner/OrphanScanRequest 传递,reconcile type-assert 到具体 *crypto.Provider 仅在装配文件 reconcile/orphan.go(装配层,类比 main)。reconcile 依赖 store/gateway/accounts/provider/crypto/config 属装配层合理。

幂等 / 崩溃安全审计: 过期清理 = status=pending 守卫条件 UPDATE(并发 settle 翻转互斥,重跑 0 行);退避/死信 = 列状态持久化,重启续跑;usage = 快照幂等覆盖;孤儿 = tx_id unique ON CONFLICT DO NOTHING;Warm = 只加不覆盖更早到期、冷却过期跳过。全部可重跑、崩溃后自愈。

Placeholder scan: 无 TODO/占位;每任务给完整 Go 代码 + 测试 + 命令。Task 5 Step 3 明确把 AddCryptoJobs 那行留到 Task 6 插入(避免 Task 5 引用未定义符号),非占位——两任务顺序落地后 main 完整。

已知取舍(记录,不阻塞 P6):

• 单实例假设:crypto 预留 Warm + orphan 扫描 + usage 快照均进程内 + 单 pay 实例语义(P3 已定「同一收款地址单实例服务」)。多实例需共享预留存储 + 扫描去重锁 + 分布式选主,超本期(设计目标单 VPS 单实例)。
• usage 快照有刷新延迟:UsageEverySec(默认 60s)窗内新付款未计入,LimitAware 可能瞬时放行略超 DailyLimit 的单——DailyLimit 是「分摊避免单账户触风控」的软阈,非硬账,秒级漂移可接受;要严格实时可改 Pick 路径直查(本期不做,避免每单 N 次 DB)。
• 抽查/孤儿扫描 TronGrid limit=50:单次拉最近 50 笔,超高频账户可能漏页——本期小体量足够,需要时加分页/游标。
• Prometheus/metrics 体系排除:告警仅 log 钩子(WithAlerter / [孤儿告警] / [对账差异] 日志),接 metrics/Telegram 属后续运维接线,不在 P6。
• 真实渠道对账单下载排除(见上决策),alipay/stripe 的渠道侧流水核对靠抽查反查 query 覆盖。

排除项(各自独立): 退款态机 P4 · 多账户路由策略 P5(已完成)· 订阅/recurring P8 · metrics/告警外发体系(运维接线)。

契约同步(pay-contract):webhook WebhookDelivery 新增 dead/next_attempt_at 列(投递侧内部状态,不影响业务方 payload);新增 orphan_payments 表(内部对账,无对外接口);config 新增 reconcile.* 段。均向后兼容(AutoMigrate 只加不删)。