fix: 停复机成功后上游状态滞后导致暂停轮询的卡到期或退款漏停
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m29s

- 停复机成功后保护窗口内(polling.stop_resume.status_grace_seconds,默认 300 秒)查卡状态返回操作前的状态视为上游滞后,不回写网络状态
- 套餐到期、退款场景不看库里的开停状态,资产无可用主套餐即停机,上游已停时按已是目标状态结束
- 迁移 000257 插入保护窗口配置

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
邹进
2026-10-26 10:41:45 +09:00
parent f6a740b9f3
commit d4fde99cc3
11 changed files with 109 additions and 5 deletions

View File

@@ -78,6 +78,20 @@ type Service struct {
trafficCalc *TrafficCalcPolicy
// logger 记录快照写入失败等不影响观测结果的告警;未注入时不输出。
logger *zap.Logger
// commandGrace 返回停复机成功后的状态保护窗口;未注入时不保护。
commandGrace func(ctx context.Context) time.Duration
}
// SetCommandGrace 注入停复机成功后的状态保护窗口(受控系统配置 polling.stop_resume.status_grace_seconds)。
func (s *Service) SetCommandGrace(grace func(ctx context.Context) time.Duration) {
s.commandGrace = grace
}
func (s *Service) currentCommandGrace(ctx context.Context) time.Duration {
if s == nil || s.commandGrace == nil {
return 0
}
return s.commandGrace(ctx)
}
// SetTrafficCalcPolicy 注入卡流量计量切换策略与告警日志。

View File

@@ -3,6 +3,7 @@ package cardobservation
import (
"context"
"github.com/break/junhong_cmp_fiber/internal/infrastructure/pollingpause"
"go.uber.org/zap"
"strconv"
"time"
@@ -47,11 +48,17 @@ func (s *Service) ApplyNetworkObservation(ctx context.Context, observation domai
auditedCard = &card
nextDecision, decisionErr := domain.ApplyNetwork(domain.CardNetworkSnapshot{
CardID: card.ID, NetworkStatus: card.NetworkStatus, StopReason: card.StopReason,
EnablePolling: card.EnablePolling,
EnablePolling: card.EnablePolling, StoppedAt: card.StoppedAt, ResumedAt: card.ResumedAt,
CommandGrace: s.currentCommandGrace(ctx),
}, observation)
if decisionErr != nil {
return decisionErr
}
if nextDecision.StatusSuppressed {
s.warnLogger().Info("停复机成功后保护窗口内观测到相反状态,判定为上游状态滞后,不回写网络状态",
zap.Uint("card_id", card.ID), zap.Int("network_status", card.NetworkStatus),
zap.String("gateway_status", observation.GatewayStatus), zap.String("metric", "card_observation.network_suppressed"))
}
decision = nextDecision
updates := map[string]any{
"last_card_status_check_at": observation.Metadata.ObservedAt,

View File

@@ -2,6 +2,9 @@ package bootstrap
import (
"context"
"github.com/redis/go-redis/v9"
"gorm.io/gorm"
"time"
"go.uber.org/zap"
@@ -208,6 +211,7 @@ func initServices(s *stores, deps *Dependencies) *services {
trafficCalcPolicy = cardObservationApp.BuildTrafficCalcPolicy(context.Background(), deps.DB, cfg.TrafficCalc, deps.Logger)
}
cardObservationService.SetTrafficCalcPolicy(trafficCalcPolicy, deps.Logger)
cardObservationService.SetCommandGrace(commandGraceProvider(deps.DB, deps.Redis, deps.Logger))
iotCard.SetCardObservationService(cardObservationService)
iotCard.SetSpeedTierIntegrationLog(integrationlog.NewRepository(deps.DB))
seriesCoordinator := cardObservationInfra.NewSeriesCoordinator(deps.Redis)
@@ -608,3 +612,9 @@ func newCarrierService(store *postgres.CarrierStore, auditWriter *auditInfra.Wri
}
return service
}
// commandGraceProvider 返回停复机成功后的状态保护窗口(受控系统配置,API 与 Worker 共用同一份配置)。
func commandGraceProvider(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) func(ctx context.Context) time.Duration {
settings := carrierqueue.NewDefaultSettingsProvider(db, redisClient, logger)
return func(ctx context.Context) time.Duration { return settings.Current(ctx).SRStatusGrace }
}

View File

@@ -127,6 +127,7 @@ func initWorkerServices(stores *queue.WorkerStores, deps *WorkerDependencies) *q
cardObservationService.SetStateAuditWriter(iotCardAuditService)
// 运营商通道流量阈值达量判定嵌入流量观测事务:与卡流量事实同事务写周期锁与停机事件。
cardObservationService.SetChannelThresholdEvaluator(carrierThresholdService)
cardObservationService.SetCommandGrace(commandGraceProvider(deps.DB, deps.Redis, deps.Logger))
// 卡流量计量切换策略:启动时展开灰度名单为内存卡集合,API 与全部 Worker 必须使用一致配置。
if cfg := config.Get(); cfg != nil {
cardObservationService.SetTrafficCalcPolicy(

View File

@@ -2,6 +2,7 @@ package cardobservation
import (
"strings"
"time"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/errors"
@@ -22,6 +23,10 @@ type CardNetworkSnapshot struct {
NetworkStatus int
StopReason string
EnablePolling bool
// StoppedAt、ResumedAt 是本系统最近一次停机、复机成功回写的时间;CommandGrace 是停复机成功后的状态保护窗口。
StoppedAt *time.Time
ResumedAt *time.Time
CommandGrace time.Duration
}
// NetworkDecision 描述一次网络观测可安全持久化的最终值。
@@ -36,6 +41,8 @@ type NetworkDecision struct {
StopReason string
StopReasonChanged bool
StopPolling bool
// StatusSuppressed 为 true 表示观测到的状态与保护窗口内刚成功的停复机相反,判定为上游状态滞后,不回写网络状态。
StatusSuppressed bool
}
// ApplyNetwork 应用 Gateway 状态映射、运营商停机原因和风险终止卡关闭轮询规则。
@@ -57,6 +64,11 @@ func ApplyNetwork(snapshot CardNetworkSnapshot, observation NetworkObservation)
if !known {
return decision, nil
}
if status != snapshot.NetworkStatus && withinCommandGrace(snapshot, observation.Metadata.ObservedAt) {
// 停复机成功后上游状态有延迟:保护窗口内观测到操作前的状态不回写,窗口过后的检查以上游为准。
decision.StatusSuppressed = true
return decision, nil
}
decision.AfterStatus = status
decision.StatusChanged = status != snapshot.NetworkStatus
if decision.StatusChanged && status == constants.NetworkStatusOffline && snapshot.StopReason == "" &&
@@ -88,3 +100,22 @@ func MapGatewayNetworkStatus(cardStatus, extend string) (int, bool) {
func ShouldStopPollingForRisk(extend string) bool {
return extend == constants.GatewayCardExtendRiskStop || extend == constants.GatewayCardExtendCancelled
}
// withinCommandGrace 判断观测时间是否落在本系统最近一次停复机成功后的保护窗口内:
// 库里为停机时看停机时间,为开机时看复机时间。
func withinCommandGrace(snapshot CardNetworkSnapshot, observedAt time.Time) bool {
if snapshot.CommandGrace <= 0 {
return false
}
last := snapshot.ResumedAt
if snapshot.NetworkStatus == constants.NetworkStatusOffline {
last = snapshot.StoppedAt
}
if last == nil || last.IsZero() {
return false
}
if observedAt.IsZero() {
observedAt = time.Now()
}
return observedAt.Sub(*last) < snapshot.CommandGrace
}

View File

@@ -26,7 +26,7 @@ func RegisterConfigDefinitions(registry *systemconfig.Registry, logger *zap.Logg
if registry == nil {
return
}
minOne, maxSeconds, maxCount := int64(1), int64(86400*7), int64(100000)
minZero, minOne, maxSeconds, maxCount := int64(0), int64(1), int64(86400*7), int64(100000)
definitions := []systemconfig.Definition{
{Key: constants.SystemConfigPollingSchedulerMode, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeString,
DefaultValue: constants.PollingSchedulerModeLegacy, Description: "卡轮询调度模式:legacy 旧分片调度,carrier 按运营商账户调度(整体切换)",
@@ -59,6 +59,8 @@ func RegisterConfigDefinitions(registry *systemconfig.Registry, logger *zap.Logg
DefaultValue: `[600,900,1800,3600]`, Description: "停复机失败退避序列(秒,递增)", Validator: validateSequence},
{Key: constants.SystemConfigStopResumeAlertFailCount, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt,
DefaultValue: "5", Description: "停复机连续失败告警阈值", Min: &minOne, Max: &maxCount},
{Key: constants.SystemConfigStopResumeStatusGrace, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt,
DefaultValue: "300", Description: "停复机成功后的状态保护窗口(秒):窗口内查卡状态返回操作前的状态视为上游滞后,不回写;0 不保护", Min: &minZero, Max: &maxSeconds},
}
for _, definition := range definitions {
if _, exists := registry.Get(definition.Key); exists {
@@ -202,6 +204,7 @@ type Settings struct {
SRUnknownAlert int
SRBackoff []time.Duration
SRAlertFailCount int
SRStatusGrace time.Duration
}
// SRBackoffFor 返回停复机第 n 次连续失败(n 从 1 起)的退避间隔;超过序列长度取最后一项。
@@ -293,6 +296,7 @@ func DefaultSettings() Settings {
SRUnknownAlert: 3,
SRBackoff: []time.Duration{10 * time.Minute, 15 * time.Minute, 30 * time.Minute, time.Hour},
SRAlertFailCount: 5,
SRStatusGrace: 5 * time.Minute,
}
}
@@ -416,6 +420,7 @@ func (p *SettingsProvider) load(ctx context.Context) (Settings, error) {
{constants.SystemConfigPollingAccountAlertThreshold, func(v int) { s.AccountAlertLimit = v }},
{constants.SystemConfigStopResumeUnknownAlert, func(v int) { s.SRUnknownAlert = v }},
{constants.SystemConfigStopResumeAlertFailCount, func(v int) { s.SRAlertFailCount = v }},
{constants.SystemConfigStopResumeStatusGrace, func(v int) { s.SRStatusGrace = time.Duration(v) * time.Second }},
}
for _, item := range ints {
raw, err := get(item.key)

View File

@@ -29,7 +29,8 @@ type stopResumeDecision struct {
}
// decideCard 统一判断函数:暂停轮询(按场景) → 停机条件 → 复机条件。
// 暂停轮询的卡只在套餐失效类场景(套餐到期、退款)下放行无套餐停机,其余自动停复机一律不做。
// 暂停轮询的卡只在套餐失效类场景(套餐到期、退款)下放行无套餐停机,其余自动停复机一律不做;
// 套餐失效类场景不看库里的开停状态,资产无可用主套餐即停机。
// 设备维度原因(无套餐、流量耗尽)与绑定设备的复机返回 DeviceID,由调用方按设备处理。通道阈值锁由复机执行方判断。
func (s *StopResumeService) decideCard(ctx context.Context, card *model.IotCard, scenes []string) (stopResumeDecision, error) {
suspended, err := s.autoControlSuspended(ctx, card)
@@ -40,6 +41,24 @@ func (s *StopResumeService) decideCard(ctx context.Context, card *model.IotCard,
if suspended && !allowSuspendedStop {
return stopResumeDecision{}, nil
}
if allowSuspendedStop {
// 套餐失效类场景(套餐到期、退款):不看库里的开停状态,资产无可用主套餐即停机。
// 停复机成功后上游状态有延迟,库里的开停状态可能是旧值;上游已停时调用按已是目标状态结束。
reasons, err := s.checkStopReasons(ctx, card)
if err != nil {
return stopResumeDecision{}, err
}
if containsStopReason(reasons, constants.StopReasonNoPackage) {
decision := stopResumeDecision{Action: constants.StopResumeActionStop, Reason: constants.StopReasonNoPackage, Reasons: reasons}
if !card.IsStandalone {
decision.DeviceID = s.boundDeviceID(ctx, card)
}
return decision, nil
}
if suspended {
return stopResumeDecision{}, nil
}
}
switch card.NetworkStatus {
case constants.NetworkStatusOnline:
reasons, err := s.checkStopReasons(ctx, card)

View File

@@ -373,8 +373,14 @@ func (s *StopResumeService) stopDeviceCards(ctx context.Context, deviceID uint,
}
defer s.redis.Del(ctx, lockKey)
// 查询设备下所有在线卡
cards, err := s.iotCardStore.ListByDeviceIDAndNetworkStatus(ctx, deviceID, constants.NetworkStatusOnline)
// 查询设备下在线卡;套餐失效类场景(套餐到期、退款)不看库里的开停状态(可能因上游延迟是旧值),设备下全部卡都停。
var cards []*model.IotCard
var err error
if stopReason == constants.StopReasonNoPackage && constants.StopResumeSceneAllowsSuspendedStop(stopResumeScenes(ctx)) {
cards, err = s.deviceCards(ctx, deviceID)
} else {
cards, err = s.iotCardStore.ListByDeviceIDAndNetworkStatus(ctx, deviceID, constants.NetworkStatusOnline)
}
if err != nil {
s.logger.Error("查询设备在线卡失败",
zap.Uint("device_id", deviceID), zap.Error(err))