fix: 停复机指令队列审查问题

- 开关开启但分发器心跳缺失时回退为当场执行，避免指令无人消费
- 待确认指令先查上游状态再判断；执行期间被覆盖时把待确认带到新指令
- 同方向失败退避中重复写入不提前执行时间
- 领取容量把执行上限与本秒剩余 QPS 分开计算
- 队列复机共用单卡复机锁；执行上下文取消后按结果未知确认
- 消费者并行停止；读取最近调用时间失败时不覆盖；成功但终结日志失败按成功处理

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/polling/carrierqueue/sr_dispatch.go

@@ -42,29 +42,30 @@ func (d *Dispatcher) stopResumeLease() time.Duration {
 // dispatchStopResume 在卡任务之前领取账户停复机指令：领取数 = min(停复机所在限流组本秒剩余, 执行上限 − 执行中)。
 func (d *Dispatcher) dispatchStopResume(ctx context.Context, carrierID uint) {
 	params := d.ParamsFor(carrierID)
-	capacity := params.SRInflightLimit
+	limit := params.SRInflightLimit
 	processes := d.cfg.SubscriberProcesses
 	if processes <= 0 {
 		processes = 1
 	}
-	if consume := params.SRMaxInFlight * processes; consume > 0 && capacity > consume {
-		capacity = consume
+	if consume := params.SRMaxInFlight * processes; consume > 0 && limit > consume {
+		limit = consume
 	}
+	budget := -1
 	if d.budgets != nil {
 		ops := []string{constants.GatewayOpStop, constants.GatewayOpStart}
 		if groups, _, limited := d.budgets.Budget(ctx, carrierID, ops); limited {
 			for _, remaining := range groups {
-				if remaining < capacity {
-					capacity = remaining
+				if budget < 0 || remaining < budget {
+					budget = remaining
 				}
 			}
 		}
 	}
-	if capacity <= 0 {
+	if limit <= 0 || budget == 0 {
 		return
 	}
 	claimed, expired, err := d.store.ClaimStopResume(ctx, SRClaimInput{
-		CarrierID: carrierID, Lease: d.stopResumeLease(), Capacity: capacity, MinInterval: params.SRMinInterval,
+		CarrierID: carrierID, Lease: d.stopResumeLease(), InflightLimit: limit, Budget: budget, MinInterval: params.SRMinInterval,
 		TokenPrefix: d.instanceID + ":sr:" + strconv.FormatInt(time.Now().UnixNano(), 36),
 	})
 	if err != nil {

internal/polling/carrierqueue/sr_queue.go

@@ -1,11 +1,29 @@
 package carrierqueue
 
-import "context"
+import (
+	"context"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/break/junhong_cmp_fiber/pkg/constants"
+)
+
+// heartbeatFreshness 是分发器心跳有效期：超过该时长没有心跳视为指令无人消费。
+const heartbeatFreshness = 30 * time.Second
+
+// heartbeatCacheTTL 是进程内缓存心跳判断的时长，避免每次请求评估都读 Redis。
+const heartbeatCacheTTL = 3 * time.Second
 
 // CommandQueue 是停复机指令队列写入端：开关读取受控系统配置，写入走共享缓存 Lua。
 type CommandQueue struct {
 	store    *Store
 	settings *SettingsProvider
+
+	mu        sync.Mutex
+	aliveAt   time.Time
+	alive     bool
+	checkedAt time.Time
 }
 
 // NewCommandQueue 创建停复机指令队列写入端。
@@ -13,9 +31,34 @@ func NewCommandQueue(store *Store, settings *SettingsProvider) *CommandQueue {
 	return &CommandQueue{store: store, settings: settings}
 }
 
-// Enabled 返回停复机出口开关 polling.stop_resume.queue_enabled。
+// Enabled 返回是否经指令队列执行：停复机出口开关开启，且分发器在心跳有效期内运行（有消费方）。
+// 开关开启但没有分发器时回退为各入口当场执行，避免指令写入后无人消费。
 func (q *CommandQueue) Enabled(ctx context.Context) bool {
-	return q != nil && q.store != nil && q.settings.Current(ctx).StopResumeQueueEnabled
+	if q == nil || q.store == nil || !q.settings.Current(ctx).StopResumeQueueEnabled {
+		return false
+	}
+	return q.dispatcherAlive(ctx)
+}
+
+func (q *CommandQueue) dispatcherAlive(ctx context.Context) bool {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if time.Since(q.checkedAt) < heartbeatCacheTTL {
+		return q.alive
+	}
+	q.checkedAt = time.Now()
+	beats, err := q.store.redis.HGetAll(ctx, constants.RedisCarrierSchedulerHeartbeatKey).Result()
+	if err != nil {
+		return q.alive
+	}
+	q.alive = false
+	for _, raw := range beats {
+		if ms, convErr := strconv.ParseInt(raw, 10, 64); convErr == nil && time.Since(time.UnixMilli(ms)) < heartbeatFreshness {
+			q.alive = true
+			break
+		}
+	}
+	return q.alive
 }
 
 // WriteCommand 写入或覆盖卡的停复机指令。

internal/polling/carrierqueue/sr_scripts.go

@@ -19,7 +19,7 @@ const (
 )
 
 // srWriteLua 写入或覆盖停复机指令：方向相反时清零失败与待确认，递增版本，放入账户停复机队列。
-// 指令执行中（存在处理中标记）时只更新指令内容与版本，由收尾按版本保留；待确认中且方向不变时保持确认时间。
+// 指令执行中（存在处理中标记）时只更新指令内容与版本，由收尾按版本保留；待确认中、或同方向失败退避中时保持原执行时间。
 //
 // KEYS[1] 指令 Hash；KEYS[2] 账户停复机队列；KEYS[3] 处理中标记。
 // ARGV：action、reason、scenes、source、carrier、now_ms、card_id。返回新版本号。
@@ -31,7 +31,9 @@ end
 redis.call('HSET', KEYS[1], 'action', ARGV[1], 'reason', ARGV[2], 'scenes', ARGV[3], 'source', ARGV[4], 'carrier', ARGV[5], 'at', ARGV[6])
 local ver = redis.call('HINCRBY', KEYS[1], 'ver', 1)
 if redis.call('EXISTS', KEYS[3]) == 1 then return ver end
-if redis.call('HGET', KEYS[1], 'pending') == '1' then
+local backingOff = prev == ARGV[1] and tonumber(redis.call('HGET', KEYS[1], 'fail') or '0') > 0
+if redis.call('HGET', KEYS[1], 'pending') == '1' or backingOff then
+  -- 待确认或同方向失败退避中：不提前执行时间（ZADD NX 只补缺失）。
   redis.call('ZADD', KEYS[2], 'NX', tonumber(ARGV[6]), ARGV[7])
 else
   redis.call('ZADD', KEYS[2], 'LT', tonumber(ARGV[6]), ARGV[7])
@@ -43,7 +45,7 @@ return ver
 // 待确认的指令只查状态，不受最小间隔约束。指令已不存在的成员直接移出队列。
 //
 // KEYS[1] 账户停复机队列；KEYS[2] 账户停复机执行中计数。
-// ARGV：now_ms、lease_ms、容量（执行上限与剩余 QPS 的较小值）、token 前缀、单卡最小间隔 ms。
+// ARGV：now_ms、lease_ms、执行上限、token 前缀、单卡最小间隔 ms、本秒剩余 QPS（-1 表示不按 QPS 限制）。
 // 返回 {过期数, card_id, token, ver, action, pending, ...}。
 const srClaimLua = `
 local now = tonumber(ARGV[1])
@@ -51,9 +53,11 @@ local lease = tonumber(ARGV[2])
 local limit = tonumber(ARGV[3])
 local prefix = ARGV[4]
 local minInterval = tonumber(ARGV[5])
+local budget = tonumber(ARGV[6])
 local expired = redis.call('ZREMRANGEBYSCORE', KEYS[2], '-inf', now)
 local result = {expired}
 local capacity = limit - redis.call('ZCARD', KEYS[2])
+if budget >= 0 and budget < capacity then capacity = budget end
 if capacity <= 0 then return result end
 local ids = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', now, 'LIMIT', 0, capacity * 2 + 10)
 local taken = 0
@@ -99,7 +103,8 @@ return result
 // 否则 done=1 删除指令并移出队列，done=0 写入字段并按 next_ms 重新排队（返回 1）。
 //
 // KEYS[1] 指令 Hash；KEYS[2] 处理中标记；KEYS[3] 执行权；KEYS[4] 执行中计数；KEYS[5] 账户停复机队列。
-// ARGV：token、card_id、领取时版本、done(1/0)、next_ms、now_ms、set 字段对数 n、字段1、值1、…、删除字段…。
+// ARGV：token、card_id、领取时版本、done(1/0)、next_ms、now_ms、本次结果未知(1/0)、set 字段对数 n、字段1、值1、…、删除字段…。
+// 版本已变大且本次结果未知时，把待确认带到新指令上：新指令执行前先查状态，避免上游已生效而库里未知。
 const srFinishLua = `
 if redis.call('GET', KEYS[2]) ~= ARGV[1] then return 0 end
 redis.call('DEL', KEYS[2])
@@ -108,6 +113,7 @@ redis.call('ZREM', KEYS[4], ARGV[1])
 local cardId = ARGV[2]
 if (redis.call('HGET', KEYS[1], 'ver') or '0') ~= ARGV[3] then
   if redis.call('EXISTS', KEYS[1]) == 1 then
+    if ARGV[7] == '1' then redis.call('HSET', KEYS[1], 'pending', '1') end
     redis.call('ZADD', KEYS[5], tonumber(ARGV[6]), cardId)
   else
     redis.call('ZREM', KEYS[5], cardId)
@@ -119,8 +125,8 @@ if ARGV[4] == '1' then
   redis.call('ZREM', KEYS[5], cardId)
   return 1
 end
-local n = tonumber(ARGV[7])
-local idx = 8
+local n = tonumber(ARGV[8])
+local idx = 9
 for i = 1, n do
   redis.call('HSET', KEYS[1], ARGV[idx], ARGV[idx + 1])
   idx = idx + 2

internal/polling/carrierqueue/sr_store.go

@@ -56,11 +56,12 @@ func (s *Store) WriteCommand(ctx context.Context, cmd StopResumeCommand) (int64,
 
 // SRClaimInput 是一次账户停复机领取的参数。
 type SRClaimInput struct {
-	CarrierID   uint
-	Lease       time.Duration
-	Capacity    int // 执行上限与剩余 QPS 的较小值
-	TokenPrefix string
-	MinInterval time.Duration
+	CarrierID     uint
+	Lease         time.Duration
+	InflightLimit int // 账户停复机执行上限（含执行中）
+	Budget        int // 停复机所在限流组本秒剩余；-1 表示不按 QPS 限制
+	TokenPrefix   string
+	MinInterval   time.Duration
 }
 
 // SRClaimed 是一条被领取的停复机指令。
@@ -76,7 +77,7 @@ type SRClaimed struct {
 func (s *Store) ClaimStopResume(ctx context.Context, input SRClaimInput) ([]SRClaimed, int64, error) {
 	raw, err := srClaimScript.Run(ctx, s.redis,
 		[]string{constants.RedisCarrierStopResumeQueueKey(input.CarrierID), constants.RedisCarrierStopResumeRunningKey(input.CarrierID)},
-		time.Now().UnixMilli(), input.Lease.Milliseconds(), input.Capacity, input.TokenPrefix, input.MinInterval.Milliseconds(),
+		time.Now().UnixMilli(), input.Lease.Milliseconds(), input.InflightLimit, input.TokenPrefix, input.MinInterval.Milliseconds(), input.Budget,
 	).Slice()
 	if err != nil || len(raw) == 0 {
 		return nil, 0, err
@@ -105,6 +106,7 @@ type SRFinish struct {
 	Token     string
 	Version   int64
 	Done      bool // true 结束并删除指令
+	Unknown   bool // 本次结果未知：版本已变大时把待确认带到新指令
 	NextAt    time.Time
 	Set       map[string]string
 	Del       []string
@@ -121,7 +123,11 @@ func (s *Store) FinishStopResume(ctx context.Context, finish SRFinish) (int64, e
 	if next.IsZero() || next.Before(now) {
 		next = now
 	}
-	args := []any{finish.Token, finish.CardID, finish.Version, done, next.UnixMilli(), now.UnixMilli(), len(finish.Set)}
+	unknown := "0"
+	if finish.Unknown {
+		unknown = "1"
+	}
+	args := []any{finish.Token, finish.CardID, finish.Version, done, next.UnixMilli(), now.UnixMilli(), unknown, len(finish.Set)}
 	for field, value := range finish.Set {
 		args = append(args, field, value)
 	}