diff --git a/config/config.go b/config/config.go index 85f6f96..071c1a9 100644 --- a/config/config.go +++ b/config/config.go @@ -102,7 +102,7 @@ type QuerySyncConfig struct { // ReconcileConfig v2 后台守护/对账周期任务开关与间隔。 type ReconcileConfig struct { Enabled bool `mapstructure:"enabled"` - OrderTTLMin int `mapstructure:"order_ttl_min"` // pending 订单存活 TTL(分钟),超则关闭 + OrderTTLMin int `mapstructure:"order_ttl_min"` // pending 订单存活 TTL(分钟),超则关闭;须 ≫ crypto 支付窗 orderTTL(15min,internal/provider/crypto/crypto.go)——否则在途 crypto 支付可能被订单级过期误杀 ExpireEverySec int `mapstructure:"expire_every_sec"` // 过期清理间隔 SyncEverySec int `mapstructure:"sync_every_sec"` // 查单对账间隔 UsageEverySec int `mapstructure:"usage_every_sec"` // 用量快照刷新间隔 @@ -114,6 +114,10 @@ type ReconcileConfig struct { // 本期归到本任务(对账主体)一起装配,brief 原表未列。 RefundApplyEverySec int `mapstructure:"refund_apply_every_sec"` // 退款修复扫描 + 卡滞告警 共用间隔 RefundStuckWarnMin int `mapstructure:"refund_stuck_warn_min"` // 退款卡滞 processing/manual_pending 告警阈值(分钟) + // RefundApplyLookbackMin 退款修复扫描的候选回溯窗(分钟):只扫 completed_at/updated_at + // 落在 [now-lookback, now] 内的候选,防 LIMIT 尾部饥饿(见 store.ListDistinctOutTradeNosByStatusSince + // / ListOrdersByStatusSince)。默认 48 小时——自愈针对近期崩溃窗口,历史一致性由已收敛状态保证。 + RefundApplyLookbackMin int `mapstructure:"refund_apply_lookback_min"` } var C Config @@ -159,6 +163,7 @@ func Load() { viper.SetDefault("reconcile.orphan_window_min", 180) viper.SetDefault("reconcile.refund_apply_every_sec", 300) viper.SetDefault("reconcile.refund_stuck_warn_min", 30) + viper.SetDefault("reconcile.refund_apply_lookback_min", 48*60) if err := viper.ReadInConfig(); err != nil { log.Println("[config] 未找到 config.yaml，使用默认值 + 环境变量") diff --git a/internal/reconcile/refund_apply.go b/internal/reconcile/refund_apply.go index fbfb561..0cd1879 100644 --- a/internal/reconcile/refund_apply.go +++ b/internal/reconcile/refund_apply.go @@ -23,14 +23,20 @@ import ( // 候选订单 = distinct(有 succeeded 退款的订单) ∪ (当前处于 refunding/partially_refunded // 态的订单):前者直接命中"退款成功但订单未跟上"的崩溃窗口;后者兜住"已在退款流程 // 中、但后续又有退款 succeeded 未被重算"的情形。 -func RefundApplyTask(orders *store.OrderStore, refunds *store.RefundStore, limit int) func(ctx context.Context) error { +// +// 两路查询都带 lookback 回溯窗(since = now-lookback),防「LIMIT 尾部饥饿」:succeeded +// 退款历史只增不减、partially_refunded 订单长期驻留,若不设窗口按 ASC 取前 limit 条, +// 超过 limit 后最新的崩溃窗口(本扫描要自愈的目标)会被挤出候选、永远扫不到。lookback +// 只覆盖近期崩溃窗口——历史一致性由已收敛状态保证,不需要扫全表。 +func RefundApplyTask(orders *store.OrderStore, refunds *store.RefundStore, lookback time.Duration, now func() time.Time, limit int) func(ctx context.Context) error { return func(ctx context.Context) error { - succeededNos, err := refunds.ListDistinctOutTradeNosByStatus(model.RefundSucceeded, limit) + since := now().Add(-lookback) + succeededNos, err := refunds.ListDistinctOutTradeNosByStatusSince(model.RefundSucceeded, since, limit) if err != nil { return err } - refundingOrders, err := orders.ListOrdersByStatus( - []model.OrderStatusV2{model.OrderRefundingV2, model.OrderPartRefundedV2}, limit) + refundingOrders, err := orders.ListOrdersByStatusSince( + []model.OrderStatusV2{model.OrderRefundingV2, model.OrderPartRefundedV2}, since, limit) if err != nil { return err } diff --git a/internal/reconcile/refund_apply_test.go b/internal/reconcile/refund_apply_test.go index 8a3c752..a85e4bd 100644 --- a/internal/reconcile/refund_apply_test.go +++ b/internal/reconcile/refund_apply_test.go @@ -25,12 +25,13 @@ func TestRefundApplyTaskSelfHealsStuckPaidOrder(t *testing.T) { if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "STUCK-1", AmountMinor: 10000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil { t.Fatal(err) } + completedAt := time.Now() if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-stuck-1", OutTradeNo: "STUCK-1", AttemptProviderRef: "STUCK-1", - AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business"}); err != nil { + AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &completedAt}); err != nil { t.Fatal(err) } - task := reconcile.RefundApplyTask(orders, refunds, 200) + task := reconcile.RefundApplyTask(orders, refunds, 48*time.Hour, time.Now, 200) if err := task(context.Background()); err != nil { t.Fatalf("task: %v", err) } @@ -51,12 +52,13 @@ func TestRefundApplyTaskPartialStaysPartial(t *testing.T) { if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "STUCK-2", AmountMinor: 10000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil { t.Fatal(err) } + completedAt := time.Now() if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-stuck-2", OutTradeNo: "STUCK-2", AttemptProviderRef: "STUCK-2", - AmountMinor: 4000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business"}); err != nil { + AmountMinor: 4000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &completedAt}); err != nil { t.Fatal(err) } - task := reconcile.RefundApplyTask(orders, refunds, 200) + task := reconcile.RefundApplyTask(orders, refunds, 48*time.Hour, time.Now, 200) if err := task(context.Background()); err != nil { t.Fatalf("task: %v", err) } @@ -76,12 +78,13 @@ func TestRefundApplyTaskIdempotentNoOpOnRerun(t *testing.T) { if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "STUCK-3", AmountMinor: 5000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil { t.Fatal(err) } + completedAt := time.Now() if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-stuck-3", OutTradeNo: "STUCK-3", AttemptProviderRef: "STUCK-3", - AmountMinor: 5000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business"}); err != nil { + AmountMinor: 5000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &completedAt}); err != nil { t.Fatal(err) } - task := reconcile.RefundApplyTask(orders, refunds, 200) + task := reconcile.RefundApplyTask(orders, refunds, 48*time.Hour, time.Now, 200) for i := 0; i < 3; i++ { if err := task(context.Background()); err != nil { t.Fatalf("run %d: %v", i, err) @@ -109,7 +112,7 @@ func TestRefundApplyTaskConsistentOrderUntouched(t *testing.T) { t.Fatal(err) } - task := reconcile.RefundApplyTask(orders, refunds, 200) + task := reconcile.RefundApplyTask(orders, refunds, 48*time.Hour, time.Now, 200) if err := task(context.Background()); err != nil { t.Fatalf("task: %v", err) } @@ -119,6 +122,100 @@ func TestRefundApplyTaskConsistentOrderUntouched(t *testing.T) { } } +// TestRefundApplyTaskLookbackWindowExcludesStaleSucceeded 钉住 LIMIT 尾部饥饿修复: +// succeeded 退款侧的候选查询带回溯窗,completed_at 超窗的行不再入候选(窗外订单卡在 +// paid 不被自愈),只有窗内的才被扫到并自愈。 +func TestRefundApplyTaskLookbackWindowExcludesStaleSucceeded(t *testing.T) { + db := model.OpenTestDB(t) + orders := store.NewOrderStore(db) + refunds := store.NewRefundStore(db) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + lookback := 48 * time.Hour + + if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "WIN-OLD", AmountMinor: 10000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil { + t.Fatal(err) + } + staleCompleted := now.Add(-lookback - time.Hour) // 窗外 + if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-win-old", OutTradeNo: "WIN-OLD", AttemptProviderRef: "WIN-OLD", + AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &staleCompleted}); err != nil { + t.Fatal(err) + } + + if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "WIN-NEW", AmountMinor: 10000, Currency: "CNY", Status: model.OrderPaidV2}); err != nil { + t.Fatal(err) + } + freshCompleted := now.Add(-time.Hour) // 窗内 + if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-win-new", OutTradeNo: "WIN-NEW", AttemptProviderRef: "WIN-NEW", + AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &freshCompleted}); err != nil { + t.Fatal(err) + } + + task := reconcile.RefundApplyTask(orders, refunds, lookback, func() time.Time { return now }, 200) + if err := task(context.Background()); err != nil { + t.Fatalf("task: %v", err) + } + + oldOrder, err := orders.GetOrder("WIN-OLD") + if err != nil || oldOrder.Status != model.OrderPaidV2 { + t.Fatalf("窗外行不应入候选、不应被自愈, got %+v, err=%v", oldOrder, err) + } + newOrder, err := orders.GetOrder("WIN-NEW") + if err != nil || newOrder.Status != model.OrderRefundedV2 { + t.Fatalf("窗内行应被自愈为 refunded, got %+v, err=%v", newOrder, err) + } +} + +// TestRefundApplyTaskLookbackWindowExcludesStaleRefundingOrder 同上,针对 +// refunding/partially_refunded 订单侧的候选查询:updated_at 超窗的订单不再入候选 +// (即便它挂着可自愈的 succeeded 退款,只要两路查询都被窗口挡在外面就不会被扫到)。 +func TestRefundApplyTaskLookbackWindowExcludesStaleRefundingOrder(t *testing.T) { + db := model.OpenTestDB(t) + orders := store.NewOrderStore(db) + refunds := store.NewRefundStore(db) + now := time.Date(2026, 7, 10, 12, 0, 0, 0, time.UTC) + lookback := 48 * time.Hour + staleAt := now.Add(-lookback - time.Hour) // 窗外 + freshAt := now.Add(-time.Hour) // 窗内 + + // 两笔订单的 succeeded 退款 completed_at 都设窗外,隔离掉「succeeded 侧候选」这条路径 + // ——只让 refunding/partially_refunded 订单侧的时间窗单独决定是否入候选。 + if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "ORD-OLD", AmountMinor: 10000, Currency: "CNY", Status: model.OrderRefundingV2}); err != nil { + t.Fatal(err) + } + if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-ord-old", OutTradeNo: "ORD-OLD", AttemptProviderRef: "ORD-OLD", + AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &staleAt}); err != nil { + t.Fatal(err) + } + if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "ORD-OLD").Update("updated_at", staleAt).Error; err != nil { + t.Fatal(err) + } + + if err := orders.CreateOrder(&model.OrderV2{OutTradeNo: "ORD-NEW", AmountMinor: 10000, Currency: "CNY", Status: model.OrderRefundingV2}); err != nil { + t.Fatal(err) + } + if err := refunds.CreateRefund(&model.Refund{RefundID: "rf-ord-new", OutTradeNo: "ORD-NEW", AttemptProviderRef: "ORD-NEW", + AmountMinor: 10000, Currency: "CNY", Status: model.RefundSucceeded, InitiatedBy: "business", CompletedAt: &staleAt}); err != nil { + t.Fatal(err) + } + if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "ORD-NEW").Update("updated_at", freshAt).Error; err != nil { + t.Fatal(err) + } + + task := reconcile.RefundApplyTask(orders, refunds, lookback, func() time.Time { return now }, 200) + if err := task(context.Background()); err != nil { + t.Fatalf("task: %v", err) + } + + oldOrder, err := orders.GetOrder("ORD-OLD") + if err != nil || oldOrder.Status != model.OrderRefundingV2 { + t.Fatalf("窗外订单不应入候选、不应被自愈, got %+v, err=%v", oldOrder, err) + } + newOrder, err := orders.GetOrder("ORD-NEW") + if err != nil || newOrder.Status != model.OrderRefundedV2 { + t.Fatalf("窗内订单应被自愈为 refunded, got %+v, err=%v", newOrder, err) + } +} + // TestRefundStuckAlertTaskLogsWarnWithoutChangingState 卡滞 processing/manual_pending // 超阈值只应打 WARN 日志,不改任何状态(观测型)。 func TestRefundStuckAlertTaskLogsWarnWithoutChangingState(t *testing.T) { diff --git a/internal/reconcile/sync_test.go b/internal/reconcile/sync_test.go index 475aca6..010d089 100644 --- a/internal/reconcile/sync_test.go +++ b/internal/reconcile/sync_test.go @@ -49,7 +49,6 @@ func TestSyncPendingTaskSettlesViaQuery(t *testing.T) { if o.Status != model.OrderPaidV2 { t.Fatalf("查单对账后应 paid, got %v", o.Status) } - _ = time.Second } func TestPaidSpotCheckTaskRunsCleanOnConsistent(t *testing.T) { diff --git a/internal/store/order_query.go b/internal/store/order_query.go index 1c08002..22bd89b 100644 --- a/internal/store/order_query.go +++ b/internal/store/order_query.go @@ -175,6 +175,22 @@ func (s *OrderStore) ListOrdersByStatus(statuses []model.OrderStatusV2, limit in return out, nil } +// ListOrdersByStatusSince 同 ListOrdersByStatus,但加 updated_at>=since 回溯窗—— +// refunding/partially_refunded 是长期驻留态(订单进入后可能停留很久),不设窗口时 +// 会占满 limit 名额,把「近期才卡滞、需要自愈」的订单挤出候选(与 +// RefundStore.ListDistinctOutTradeNosByStatusSince 同一治法,同一 lookback 语义)。 +func (s *OrderStore) ListOrdersByStatusSince(statuses []model.OrderStatusV2, since time.Time, limit int) ([]model.OrderV2, error) { + if limit <= 0 || limit > 500 { + limit = 200 + } + var out []model.OrderV2 + if err := s.db.Where("status IN ? AND updated_at >= ?", statuses, since). + Order("id ASC").Limit(limit).Find(&out).Error; err != nil { + return nil, fmt.Errorf("store.ListOrdersByStatusSince: %w", err) + } + return out, nil +} + // ListAttemptsByChannelSince 列某渠道 created_at>=since 的 attempt(任意状态), // 供 orphan 扫描构造"已知期望金额集"(凡 pay 合法签发过的金额都不算孤儿)。 func (s *OrderStore) ListAttemptsByChannelSince(channel string, since time.Time, limit int) ([]model.Attempt, error) { diff --git a/internal/store/order_query_test.go b/internal/store/order_query_test.go index 254ecc3..82ffbd5 100644 --- a/internal/store/order_query_test.go +++ b/internal/store/order_query_test.go @@ -177,6 +177,37 @@ func TestListOrdersByStatus(t *testing.T) { } } +// TestListOrdersByStatusSinceExcludesStale 钉住回溯窗:updated_at 早于 since 的订单 +// 不应入候选(防 refunding/partially_refunded 长期驻留占满 LIMIT 名额——见 RefundApplyTask)。 +func TestListOrdersByStatusSinceExcludesStale(t *testing.T) { + db := model.OpenTestDB(t) + s := store.NewOrderStore(db) + now := time.Now() + stale := now.Add(-72 * time.Hour) + fresh := now.Add(-time.Hour) + + if err := s.CreateOrder(&model.OrderV2{OutTradeNo: "O-STALE", AmountMinor: 100, Currency: "CNY", Status: model.OrderRefundingV2}); err != nil { + t.Fatal(err) + } + if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "O-STALE").Update("updated_at", stale).Error; err != nil { + t.Fatal(err) + } + if err := s.CreateOrder(&model.OrderV2{OutTradeNo: "O-FRESH", AmountMinor: 100, Currency: "CNY", Status: model.OrderRefundingV2}); err != nil { + t.Fatal(err) + } + if err := db.Model(&model.OrderV2{}).Where("out_trade_no = ?", "O-FRESH").Update("updated_at", fresh).Error; err != nil { + t.Fatal(err) + } + + got, err := s.ListOrdersByStatusSince([]model.OrderStatusV2{model.OrderRefundingV2}, now.Add(-48*time.Hour), 50) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 1 || got[0].OutTradeNo != "O-FRESH" { + t.Fatalf("只应命中窗内的 O-FRESH, got %+v", got) + } +} + // TestListAttemptsByChannelSince 供 orphan 扫描构造"已知期望金额集":同渠道、 // created_at>=since、任意状态的 attempt(pending/paid/expired 都算——凡 pay // 合法签发过的金额都不算孤儿)。 diff --git a/internal/store/refund.go b/internal/store/refund.go index 49370be..47d4294 100644 --- a/internal/store/refund.go +++ b/internal/store/refund.go @@ -144,6 +144,27 @@ func (s *RefundStore) ListDistinctOutTradeNosByStatus(status model.RefundStatus, return out, nil } +// ListDistinctOutTradeNosByStatusSince 同 ListDistinctOutTradeNosByStatus,但加 +// completed_at>=since 回溯窗——防「LIMIT 尾部饥饿」:succeeded 退款历史只增不减, +// 若不设窗口,按 out_trade_no ASC 取前 limit 条,超过 limit 后最新出现的"退款已成但 +// 订单卡 paid"崩溃窗口(本扫描要自愈的目标)会被排在候选之外、永远扫不到。completed_at +// 是 MarkRefundStatus 翻成 succeeded/failed 时打的时间戳(即"进入该终态"的时刻),succeeded +// 是终态、之后不会再被改动,用它做窗口锚点比 updated_at 更精确。调用方(RefundApplyTask) +// 用 now-lookback 算 since,lookback 只覆盖近期崩溃窗口——历史一致性由已收敛状态保证, +// 不需要扫全表。 +func (s *RefundStore) ListDistinctOutTradeNosByStatusSince(status model.RefundStatus, since time.Time, limit int) ([]string, error) { + if limit <= 0 || limit > 500 { + limit = 200 + } + var out []string + if err := s.db.Model(&model.Refund{}).Where("status = ? AND completed_at >= ?", status, since). + Group("out_trade_no").Order("out_trade_no ASC").Limit(limit). + Pluck("out_trade_no", &out).Error; err != nil { + return nil, fmt.Errorf("store.ListDistinctOutTradeNosByStatusSince: %w", err) + } + return out, nil +} + // ListStuckRefunds 列 status 落在给定集合、且 updated_at 早于 before 的退款行,供 // 「卡滞 processing/manual_pending 退款告警」只读观测扫描用(不改状态)。updated_at // 用作「进入当前状态」的近似时刻——本表除 MarkRefundStatus/CreateRefundGuarded 外 diff --git a/internal/store/refund_test.go b/internal/store/refund_test.go index 7370a5e..61a0bfd 100644 --- a/internal/store/refund_test.go +++ b/internal/store/refund_test.go @@ -126,6 +126,29 @@ func TestListDistinctOutTradeNosByStatus(t *testing.T) { } } +// TestListDistinctOutTradeNosByStatusSinceExcludesStale 钉住回溯窗:completed_at +// 早于 since 的 succeeded 退款不应入候选(防 LIMIT 尾部饥饿——见 RefundApplyTask)。 +func TestListDistinctOutTradeNosByStatusSinceExcludesStale(t *testing.T) { + db := model.OpenTestDB(t) + rs := NewRefundStore(db) + now := time.Now() + stale := now.Add(-72 * time.Hour) + fresh := now.Add(-time.Hour) + + _ = rs.CreateRefund(&model.Refund{RefundID: "s-stale", OutTradeNo: "PAY-STALE", AmountMinor: 100, Currency: "CNY", + Status: model.RefundSucceeded, CompletedAt: &stale}) + _ = rs.CreateRefund(&model.Refund{RefundID: "s-fresh", OutTradeNo: "PAY-FRESH", AmountMinor: 100, Currency: "CNY", + Status: model.RefundSucceeded, CompletedAt: &fresh}) + + got, err := rs.ListDistinctOutTradeNosByStatusSince(model.RefundSucceeded, now.Add(-48*time.Hour), 50) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 1 || got[0] != "PAY-FRESH" { + t.Fatalf("只应命中窗内的 PAY-FRESH, got %+v", got) + } +} + // TestListStuckRefunds 供「卡滞 processing/manual_pending 退款告警」:只挑 updated_at // 早于阈值的 processing/manual_pending 行,requested/succeeded/failed 不命中。 func TestListStuckRefunds(t *testing.T) { diff --git a/main.go b/main.go index 89864bd..c52675c 100644 --- a/main.go +++ b/main.go @@ -77,7 +77,7 @@ func main() { runner.Add("paid-spotcheck", time.Duration(rc.SpotCheckEverySec)*time.Second, reconcile.PaidSpotCheckTask(orderStore, pReg, time.Duration(rc.SpotCheckWindowMin)*time.Minute, time.Now)) runner.Add("refund-apply-sweep", time.Duration(rc.RefundApplyEverySec)*time.Second, - reconcile.RefundApplyTask(orderStore, refundStore, 200)) + reconcile.RefundApplyTask(orderStore, refundStore, time.Duration(rc.RefundApplyLookbackMin)*time.Minute, time.Now, 200)) runner.Add("refund-stuck-alert", time.Duration(rc.RefundApplyEverySec)*time.Second, reconcile.RefundStuckAlertTask(refundStore, time.Duration(rc.RefundStuckWarnMin)*time.Minute, time.Now)) orphanStore := store.NewOrphanStore(db)