bugfix

internal/service/virtualpay_compensate.go

@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 	"log/slog"
+	"sync"
 	"time"
 
 	"git.lzcwy.cn/zhengchiliu/zt-closet-backend/global"
@@ -17,8 +18,20 @@ const (
 	// virtualConfirmLookbackSeconds 查单补偿的回看窗口：只处理近 2 小时内的待支付/已过期虚拟订单。
 	// 该窗口覆盖推送重试耗尽的时间（官方最多重试 15 次、间隔 2/4/8/16… 秒），又避免全表扫描。
 	virtualConfirmLookbackSeconds int64 = 2 * 60 * 60
-	// virtualConfirmBatchSize 单轮最多处理的订单数，控制单轮耗时与微信接口调用量。
-	virtualConfirmBatchSize = 50
+	// virtualConfirmBatchSize 单轮最多取回的候选订单数。
+	virtualConfirmBatchSize = 200
+	// virtualConfirmWorkers 单轮并发查询的 worker 数。
+	// 必须并发：串行查询在积压时会撞上 ctx 超时，导致尾部订单被静默跳过（看似处理了全部，实则只查了前一部分）。
+	virtualConfirmWorkers = 5
+	// virtualConfirmUnpaidTTL 「已确认未支付」负缓存的存活时间。
+	// 命中即跳过本轮查询，使积压订单让出扫描名额——这是修复「积压订单饿死已支付订单」的关键；
+	// 同时把单个订单的查询频率从每分钟一次降到每 TTL 一次，控制微信接口调用量。
+	// 代价：用户支付后最坏要等一个 TTL 才被补偿发现，需要更快收敛就把该值调小。
+	virtualConfirmUnpaidTTL = 10 * time.Minute
+	// virtualConfirmLockTTL 单轮任务的分布式锁 TTL，必须大于单轮最大耗时，否则锁提前过期会让多实例并发执行。
+	virtualConfirmLockTTL = 5 * time.Minute
+	// virtualConfirmTimeout 单轮任务的上下文超时，需给「批量查询 + 并发反查」留足余量。
+	virtualConfirmTimeout = 10 * time.Minute
 )
 
 // StartVirtualPayConfirmScheduler 启动虚拟支付查单补偿定时任务（官方「道具直购流程」的发货轮询分支）。
@@ -45,14 +58,20 @@ func StartVirtualPayConfirmScheduler() {
 	}()
 }
 
-// runVirtualPayConfirm 扫描待确认的虚拟支付订单并逐个向微信反查，单实例通过 Redis 分布式锁防重。
+// runVirtualPayConfirm 扫描待确认的虚拟支付订单并并发向微信反查，单实例通过 Redis 分布式锁防重。
+//
+// 扫描策略（避免积压饿死）：
+//  1. 按 id 升序取窗口内候选订单，用「已确认未支付」负缓存批量过滤掉最近查过的订单，
+//     被过滤的订单不再占名额，后续新订单自然会被覆盖到；
+//  2. 剩余候选用固定 worker 并发反查，避免串行耗时撞上 ctx 超时；
+//  3. 候选数打满上限时打 warning，提示可能存在积压。
 func runVirtualPayConfirm() {
 	// 虚拟支付未开通时直接返回，避免无谓的数据库扫描。
 	if !wechat.IsVirtualPayEnabled() {
 		return
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
+	ctx, cancel := context.WithTimeout(context.Background(), virtualConfirmTimeout)
 	defer cancel()
 
 	ctx, span := global.Tracer.Start(ctx, "scheduler.VirtualPayConfirm")
@@ -60,7 +79,7 @@ func runVirtualPayConfirm() {
 
 	lockKey := redismodel.KeyLockVirtualConfirm
 	lockVal := genUUID()
-	ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, time.Minute)
+	ok, lerr := redisdao.TryLock(ctx, lockKey, lockVal, virtualConfirmLockTTL)
 	if lerr != nil || !ok {
 		return
 	}
@@ -76,17 +95,65 @@ func runVirtualPayConfirm() {
 	if len(orders) == 0 {
 		return
 	}
+	if len(orders) >= virtualConfirmBatchSize {
+		slog.WarnContext(ctx, "virtual confirm backlog detected, candidates reached limit",
+			"candidates", len(orders), "limit", virtualConfirmBatchSize,
+			"lookback_seconds", virtualConfirmLookbackSeconds)
+	}
+
+	// 批量过滤负缓存：一次 MGET，避免逐单往返 Redis。
+	orderNos := make([]string, 0, len(orders))
+	for i := range orders {
+		orderNos = append(orderNos, orders[i].OrderNo)
+	}
+	unpaid, merr := redisdao.VirtualPayUnpaidBatchGet(ctx, orderNos)
+	if merr != nil {
+		// 负缓存不可用不阻断补偿：本轮退化为「逐单都查」（最坏情况等价于旧行为）。
+		slog.WarnContext(ctx, "get virtual pay unpaid cache failed, fallback to query all", "error", merr)
+		unpaid = map[string]bool{}
+	}
+
+	targets := make([]*mysqlmodel.Order, 0, len(orders))
+	for i := range orders {
+		if unpaid[orders[i].OrderNo] {
+			continue // 最近已确认未支付，让出名额给其它订单。
+		}
+		targets = append(targets, &orders[i])
+	}
+	if len(targets) == 0 {
+		return
+	}
 
 	s := NewOrderService()
-	confirmed := 0
-	for i := range orders {
-		if s.confirmVirtualOrder(ctx, &orders[i]) {
-			confirmed++
-		}
+	var (
+		mu        sync.Mutex
+		wg        sync.WaitGroup
+		confirmed int
+	)
+	sem := make(chan struct{}, virtualConfirmWorkers)
+	for _, order := range targets {
+		wg.Add(1)
+		sem <- struct{}{}
+		go func(o *mysqlmodel.Order) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			defer func() {
+				if r := recover(); r != nil {
+					slog.Error("virtual confirm worker panic", "order_no", o.OrderNo, "recover", r)
+				}
+			}()
+			if s.confirmVirtualOrder(ctx, o) {
+				mu.Lock()
+				confirmed++
+				mu.Unlock()
+			}
+		}(order)
 	}
+	wg.Wait()
+
 	if confirmed > 0 {
 		slog.InfoContext(ctx, "virtual pay confirm compensated orders",
-			"scanned", len(orders), "confirmed", confirmed)
+			"scanned", len(orders), "queried", len(targets), "confirmed", confirmed)
 	}
 }
 
@@ -104,12 +171,18 @@ func (s *OrderService) confirmVirtualOrder(ctx context.Context, order *mysqlmode
 
 	info, qerr := wechat.QueryOrder(ctx, user.WxOpenID, order.OrderNo, "")
 	if qerr != nil {
+		// 查询失败不等于「未支付」，不写负缓存，下一轮继续重试。
 		slog.ErrorContext(ctx, "virtual confirm: query order failed",
 			"order_no", order.OrderNo, "error", qerr)
 		return false
 	}
 	if !wechat.IsOrderPaidStatus(info.Order.Status) {
 		// 未支付：用户放弃或仍在支付中，保持原状态等待过期。
+		// 写入负缓存让该订单在 TTL 内让出扫描名额，避免大量未支付订单把已支付订单挤出扫描窗口。
+		if serr := redisdao.VirtualPayUnpaidSet(ctx, order.OrderNo, virtualConfirmUnpaidTTL); serr != nil {
+			slog.WarnContext(ctx, "virtual confirm: set unpaid cache failed",
+				"order_no", order.OrderNo, "error", serr)
+		}
 		return false
 	}
 	// 金额必须与下单快照一致，防止订单号错配导致错发。
@@ -138,6 +211,11 @@ func (s *OrderService) confirmVirtualOrder(ctx context.Context, order *mysqlmode
 				slog.ErrorContext(ctx, "virtual confirm: notify provide goods failed",
 					"order_no", order.OrderNo, "error", nerr)
 			}
+			// 订单已补发，清掉可能残留的负缓存。
+			if derr := redisdao.VirtualPayUnpaidDel(ctx, order.OrderNo); derr != nil {
+				slog.WarnContext(ctx, "virtual confirm: clear unpaid cache failed",
+					"order_no", order.OrderNo, "error", derr)
+			}
 			return true
 		}
 	}