diff --git a/cmd/worker/carrier_scheduler.go b/cmd/worker/carrier_scheduler.go index 6ec462fa..d146ebc1 100644 --- a/cmd/worker/carrier_scheduler.go +++ b/cmd/worker/carrier_scheduler.go @@ -289,13 +289,19 @@ const deviceEvalMaxInFlight = 10 // shutdown 优雅退出：停止订阅，宽限期内等待执行中的卡任务完成；未完成的由租约到期后重新领取。 func (c *carrierSchedulerRuntime) shutdown() { - if c.consumers != nil { - c.consumers.StopAll(c.grace()) - } - if c.srConsumers != nil { - c.srConsumers.StopAll(c.grace()) - c.devConsumers.StopAll(c.grace()) + // 三类消费者并行停止，共用同一宽限期。 + var wg sync.WaitGroup + for _, manager := range []*nsqbus.ConsumerManager{c.consumers, c.srConsumers, c.devConsumers} { + if manager == nil { + continue + } + wg.Add(1) + go func(m *nsqbus.ConsumerManager) { + defer wg.Done() + m.StopAll(c.grace()) + }(manager) } + wg.Wait() if c.producer != nil { c.producer.Stop() } diff --git a/internal/polling/carrierqueue/sr_dispatch.go b/internal/polling/carrierqueue/sr_dispatch.go index b39fefa7..7981d9f8 100644 --- a/internal/polling/carrierqueue/sr_dispatch.go +++ b/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 { diff --git a/internal/polling/carrierqueue/sr_queue.go b/internal/polling/carrierqueue/sr_queue.go index eee98b24..2d02dbed 100644 --- a/internal/polling/carrierqueue/sr_queue.go +++ b/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 写入或覆盖卡的停复机指令。 diff --git a/internal/polling/carrierqueue/sr_scripts.go b/internal/polling/carrierqueue/sr_scripts.go index c1935721..d467d893 100644 --- a/internal/polling/carrierqueue/sr_scripts.go +++ b/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 diff --git a/internal/polling/carrierqueue/sr_store.go b/internal/polling/carrierqueue/sr_store.go index 703deb9b..200bb780 100644 --- a/internal/polling/carrierqueue/sr_store.go +++ b/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) } diff --git a/internal/service/carrier/service.go b/internal/service/carrier/service.go index fcc4a32b..e1d9a1fe 100644 --- a/internal/service/carrier/service.go +++ b/internal/service/carrier/service.go @@ -238,7 +238,11 @@ func (s *Service) Update(ctx context.Context, id uint, req *dto.UpdateCarrierReq func (s *Service) requestRealnameControlEvaluation(ctx context.Context, carrierID uint, enabled bool) { evalCtx := context.WithoutCancel(ctx) go func() { - defer func() { _ = recover() }() + defer func() { + if r := recover(); r != nil { + s.logger.Error("实名管控修改后批量请求停复机评估 panic", zap.Uint("carrier_id", carrierID), zap.Any("panic", r)) + } + }() if _, err := s.realnameEvaluator.RequestRealnameControlEvaluation(evalCtx, carrierID, enabled); err != nil { s.logger.Error("实名管控修改后批量请求停复机评估失败", zap.Uint("carrier_id", carrierID), zap.Error(err)) } diff --git a/internal/service/iot_card/stop_resume_command.go b/internal/service/iot_card/stop_resume_command.go index 9617cd3d..90794be5 100644 --- a/internal/service/iot_card/stop_resume_command.go +++ b/internal/service/iot_card/stop_resume_command.go @@ -52,6 +52,9 @@ func (s *StopResumeService) DecideCommand(ctx context.Context, cardID uint, scen return CommandDecision{Card: card, Action: decision.Action, Reason: decision.Reason}, nil } +// ErrCommandBusy 表示该卡复机正在其他路径执行（持单卡复机锁），本次未调用上游。 +var ErrCommandBusy = stderrors.New("该卡复机正在执行中") + // CommandCallResult 是一次停复机网关调用的结果。 type CommandCallResult struct { // Unknown 为 true 表示结果未知：超时、连接失败，或上游已成功但本地回写失败。 @@ -66,8 +69,22 @@ func (s *StopResumeService) ExecuteCommand(ctx context.Context, card *model.IotC } if action == constants.StopResumeActionStop { outcome, err := s.stopCardAttempts(ctx, card, reason, 1) + if err != nil && outcome.Result == constants.AuditResultSuccess { + // 停机与回写已成功，只是 Integration Log 终结失败：按成功处理。 + s.logger.Warn("停机成功但终结 Integration Log 失败", zap.Uint("card_id", card.ID), zap.Error(err)) + return CommandCallResult{}, nil + } return CommandCallResult{Unknown: err != nil && outcome.Result == constants.AuditResultUnknown && !gateway.IsRateLimited(err)}, err } + // 与旧复机路径共用单卡复机锁，避免与自动续费、通道阈值等直接复机并发。 + if s.redis != nil { + lockKey := constants.RedisCardResumeLockKey(card.ID) + locked, lockErr := s.redis.SetNX(ctx, lockKey, "1", 30*time.Second).Result() + if lockErr != nil || !locked { + return CommandCallResult{}, ErrCommandBusy + } + defer s.redis.Del(context.WithoutCancel(ctx), lockKey) + } actionCode, summary := constants.AuditActionIotCardAutoStarted, "自动恢复 IoT 卡网络" attempt, integrationID, err := s.resumeCardAttempts(ctx, card, actionCode, summary, 1) if err != nil { diff --git a/internal/service/iot_card/stop_resume_request.go b/internal/service/iot_card/stop_resume_request.go index 1b32da4e..815b6f33 100644 --- a/internal/service/iot_card/stop_resume_request.go +++ b/internal/service/iot_card/stop_resume_request.go @@ -2,6 +2,7 @@ package iot_card import ( "context" + stderrors "errors" "go.uber.org/zap" @@ -62,11 +63,15 @@ func (s *StopResumeService) legacyAssetEvaluation(ctx context.Context, assetType if err != nil { return err } + var failed []error for _, card := range cards { if err := s.EvaluateAndAct(ctx, card); err != nil { - return err + failed = append(failed, err) } } + if len(failed) > 0 { + return errors.Wrap(errors.CodeInternalError, stderrors.Join(failed...), "设备停复机评估存在未成功处理的卡") + } return nil } card, err := s.iotCardStore.GetByID(ctx, assetID) diff --git a/internal/task/carrier_stop_resume_executor.go b/internal/task/carrier_stop_resume_executor.go index 971e38e5..14a5b09a 100644 --- a/internal/task/carrier_stop_resume_executor.go +++ b/internal/task/carrier_stop_resume_executor.go @@ -3,6 +3,7 @@ package task import ( "context" "encoding/json" + stderrors "errors" "strconv" "time" @@ -95,7 +96,7 @@ func (e *StopResumeExecutor) Handle(ctx context.Context, body []byte, _ func()) e.countOutcome(finishCtx, msg.CarrierID, outcome) result, err := e.store.FinishStopResume(finishCtx, carrierqueue.SRFinish{ CardID: msg.CardID, CarrierID: msg.CarrierID, Token: msg.Token, Version: msg.Version, - Done: outcome.done, NextAt: outcome.nextAt, Set: outcome.set, Del: outcome.del, + Done: outcome.done, Unknown: outcome.set[carrierqueue.SRFieldPending] == "1", NextAt: outcome.nextAt, Set: outcome.set, Del: outcome.del, }) switch { case err != nil: @@ -129,75 +130,92 @@ func (e *StopResumeExecutor) execute(ctx context.Context, msg carrierqueue.StopR if decision.Gone { return srEnded() } - if decision.Action != cmd.Action { - // 条件不再满足或已是目标状态：结束指令，不调用上游。 - e.logger.Info("停复机指令执行前重新判断不再需要，结束指令", zap.Uint("card_id", msg.CardID), - zap.String("action", cmd.Action), zap.String("decided", decision.Action), zap.Strings("scenes", cmd.Scenes)) - e.recovered(cmd, decision.Card) - return srEnded() - } card := decision.Card - reason := decision.Reason params := e.params(card.CarrierID) settings := e.settings.Current(ctx) gatewayCtx := gateway.WithRateLimitWait(ctx, 20*time.Second) - if !cmd.Pending { - return e.call(gatewayCtx, card, cmd, reason, params, settings) + // 待确认（上次结果未知）：先查上游状态再按规则判断，避免上游已生效而库里未知时直接结束指令。 + clearedPending := false + if cmd.Pending { + if outcome, settled := e.confirm(gatewayCtx, card, cmd, settings); settled { + return outcome + } + clearedPending = true } - if confirmed, next := e.confirm(gatewayCtx, card, cmd, reason, params, settings); confirmed != nil { - return *confirmed - } else if next != nil { - return *next + if decision.Action != cmd.Action { + // 条件不再满足或已是目标状态：结束指令，不调用上游。 + e.logger.Info("停复机指令执行前重新判断不再需要，结束指令", zap.Uint("card_id", msg.CardID), + zap.String("action", cmd.Action), zap.String("decided", decision.Action), zap.Strings("scenes", cmd.Scenes)) + e.recovered(cmd, card) + return srEnded() } - // 确认不是目标状态且已满足单卡最小间隔：清除待确认后重新调用。 - result := e.call(gatewayCtx, card, cmd, reason, params, settings) - if !result.done && result.set[carrierqueue.SRFieldPending] != "1" { + if clearedPending { + // 确认不是目标状态：未到单卡最小间隔则推后，否则清除待确认后重新调用。 + last, lastErr := e.store.LastStopResumeCall(ctx, card.ID) + if allowed := last.Add(params.SRMinInterval); lastErr != nil || (!last.IsZero() && allowed.After(time.Now())) { + if lastErr != nil { + allowed = time.Now().Add(time.Minute) + } + retry := srRetry(allowed) + retry.del = []string{carrierqueue.SRFieldPending} + return retry + } + } + result := e.call(gatewayCtx, card, cmd, decision.Reason, params, settings) + if clearedPending && !result.done && result.set[carrierqueue.SRFieldPending] != "1" { result.del = append(result.del, carrierqueue.SRFieldPending) } return result } -// confirm 处理待确认指令：查一次卡状态。已是目标状态返回结束；查询失败保持待确认按退避重查； -// 不是目标状态时清除待确认，未到单卡最小间隔则推后，否则返回 (nil, nil) 由调用方重新发起停复机。 -func (e *StopResumeExecutor) confirm(ctx context.Context, card *model.IotCard, cmd carrierqueue.CommandState, reason string, - params carrierqueue.CarrierParams, settings carrierqueue.Settings) (*srOutcome, *srOutcome) { +// confirm 处理待确认指令：查一次卡状态（受限流、不受单卡最小间隔约束）。 +// 已是目标状态：按成功回写并结束；若此时规则要求相反动作，写入新指令（版本变大，收尾保留新指令）。 +// 查询失败：保持待确认按退避重查，连续未知达阈值告警，不发起停复机。 +// 不是目标状态：返回 settled=false，由调用方按最新判断继续。 +func (e *StopResumeExecutor) confirm(ctx context.Context, card *model.IotCard, cmd carrierqueue.CommandState, + settings carrierqueue.Settings) (srOutcome, bool) { status, known, err := e.service.QueryCommandCardStatus(ctx, card) if gateway.IsRateLimited(err) { - retry := srRetry(time.Now().Add(stopResumeRateLimitedDelay)) - return nil, &retry + return srRetry(time.Now().Add(stopResumeRateLimitedDelay)), true } if err != nil || !known { - retry := srRetry(time.Now().Add(settings.SRBackoffFor(cmd.Unknown))) + unknown := cmd.Unknown + 1 + retry := srRetry(time.Now().Add(settings.SRBackoffFor(unknown))) + retry.set[carrierqueue.SRFieldUnknown] = strconv.Itoa(unknown) + message := "确认查询未得到明确状态" if err != nil { - retry.set[carrierqueue.SRFieldError] = truncateRunes(constants.StopResumeFailureUnknown+"|确认查询失败: "+err.Error(), 300) + message = "确认查询失败: " + err.Error() } - return nil, &retry - } - if status == targetNetworkStatus(cmd.Action) { - if confirmErr := e.service.ConfirmCommandState(ctx, card, cmd.Action, reason); confirmErr != nil { - e.logger.Warn("确认已是目标状态但回写失败，稍后再确认", zap.Uint("card_id", card.ID), zap.Error(confirmErr)) - retry := srRetry(time.Now().Add(time.Minute)) - return nil, &retry + retry.set[carrierqueue.SRFieldError] = truncateRunes(constants.StopResumeFailureUnknown+"|"+message, 300) + if unknown >= settings.SRUnknownAlert { + e.reportFailure(card, cmd, constants.StopResumeFailureUnknown, message, unknown, cmd.WrittenAt) } - e.logger.Info("停复机结果未知，确认查询已是目标状态，按成功结束", zap.Uint("card_id", card.ID), zap.String("action", cmd.Action)) - e.recovered(cmd, card) - done := srDone() - return &done, nil + return retry, true } - last, _ := e.store.LastStopResumeCall(ctx, card.ID) - if allowed := last.Add(params.SRMinInterval); !last.IsZero() && allowed.After(time.Now()) { - retry := srRetry(allowed) - retry.del = []string{carrierqueue.SRFieldPending} - return nil, &retry + if status != targetNetworkStatus(cmd.Action) { + return srOutcome{}, false } - return nil, nil + if confirmErr := e.service.ConfirmCommandState(ctx, card, cmd.Action, cmd.Reason); confirmErr != nil { + e.logger.Warn("确认已是目标状态但回写失败，稍后再确认", zap.Uint("card_id", card.ID), zap.Error(confirmErr)) + return srRetry(time.Now().Add(time.Minute)), true + } + e.logger.Info("停复机结果未知，确认查询已是目标状态，按成功结束", zap.Uint("card_id", card.ID), zap.String("action", cmd.Action)) + e.recovered(cmd, card) + if next, decideErr := e.service.DecideCommand(ctx, card.ID, cmd.Scenes); decideErr == nil && next.Card != nil && + next.Action != "" && next.Action != cmd.Action { + if _, writeErr := e.store.WriteCommand(ctx, carrierqueue.StopResumeCommand{CardID: card.ID, CarrierID: card.CarrierID, + Action: next.Action, Reason: next.Reason, Scenes: cmd.Scenes, Source: "confirm"}); writeErr != nil { + e.logger.Warn("确认后按规则写入相反指令失败", zap.Uint("card_id", card.ID), zap.Error(writeErr)) + } + } + return srDone(), true } // call 调用一次停机或复机并按结果分类处理。 func (e *StopResumeExecutor) call(ctx context.Context, card *model.IotCard, cmd carrierqueue.CommandState, reason string, params carrierqueue.CarrierParams, settings carrierqueue.Settings) srOutcome { - previousCall, _ := e.store.LastStopResumeCall(ctx, card.ID) + previousCall, previousErr := e.store.LastStopResumeCall(ctx, card.ID) now := time.Now() if err := e.store.MarkStopResumeCall(ctx, card.ID, now, 2*params.SRMinInterval); err != nil { return srRetry(now.Add(time.Minute)) @@ -209,13 +227,16 @@ func (e *StopResumeExecutor) call(ctx context.Context, card *model.IotCard, cmd e.recovered(cmd, card) return srDone() } - if gateway.IsRateLimited(err) { - // 本账户限流：请求未发出，恢复最近调用时间，短延迟重排，不计失败。 - e.restoreLastCall(ctx, card.ID, previousCall, params) + if gateway.IsRateLimited(err) || stderrors.Is(err, iot_card_svc.ErrCommandBusy) { + // 本账户限流或该卡复机正在其他路径执行：请求未发出，恢复最近调用时间（读取失败时保守保留），短延迟重排，不计失败。 + if previousErr == nil { + e.restoreLastCall(ctx, card.ID, previousCall, params) + } return srRetry(time.Now().Add(stopResumeRateLimitedDelay)) } classification := e.classify(ctx, cmd.Action, err) - if result.Unknown || classification.MaybeAccepted { + // 执行上下文已取消（退出或超时）时请求可能已发出：按结果未知确认，不计失败。 + if result.Unknown || classification.MaybeAccepted || ctx.Err() != nil { return e.unknown(card, cmd, err, params, settings) } message := truncateRunes(classification.Category+"|"+err.Error(), 300)