diff --git a/internal/application/cardobservation/apply.go b/internal/application/cardobservation/apply.go index 0623d415..6f215334 100644 --- a/internal/application/cardobservation/apply.go +++ b/internal/application/cardobservation/apply.go @@ -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 注入卡流量计量切换策略与告警日志。 diff --git a/internal/application/cardobservation/network.go b/internal/application/cardobservation/network.go index 1789278a..0fbaea1d 100644 --- a/internal/application/cardobservation/network.go +++ b/internal/application/cardobservation/network.go @@ -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, diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index 95821afa..cb496f4b 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -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 } +} diff --git a/internal/bootstrap/worker_services.go b/internal/bootstrap/worker_services.go index 56592495..465c5122 100644 --- a/internal/bootstrap/worker_services.go +++ b/internal/bootstrap/worker_services.go @@ -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( diff --git a/internal/domain/cardobservation/network.go b/internal/domain/cardobservation/network.go index 2fd8a00a..91d995ba 100644 --- a/internal/domain/cardobservation/network.go +++ b/internal/domain/cardobservation/network.go @@ -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 +} diff --git a/internal/polling/carrierqueue/settings.go b/internal/polling/carrierqueue/settings.go index 1b44f524..9fe340ea 100644 --- a/internal/polling/carrierqueue/settings.go +++ b/internal/polling/carrierqueue/settings.go @@ -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) diff --git a/internal/service/iot_card/stop_resume_decide.go b/internal/service/iot_card/stop_resume_decide.go index 56571184..e4a17747 100644 --- a/internal/service/iot_card/stop_resume_decide.go +++ b/internal/service/iot_card/stop_resume_decide.go @@ -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) diff --git a/internal/service/iot_card/stop_resume_service.go b/internal/service/iot_card/stop_resume_service.go index a3bca40e..15357acc 100644 --- a/internal/service/iot_card/stop_resume_service.go +++ b/internal/service/iot_card/stop_resume_service.go @@ -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)) diff --git a/migrations/000257_add_stop_resume_status_grace_config.down.sql b/migrations/000257_add_stop_resume_status_grace_config.down.sql new file mode 100644 index 00000000..2c6dd907 --- /dev/null +++ b/migrations/000257_add_stop_resume_status_grace_config.down.sql @@ -0,0 +1,2 @@ +-- 回滚停复机成功后的状态保护窗口配置。 +DELETE FROM tb_system_config WHERE config_key = 'polling.stop_resume.status_grace_seconds'; diff --git a/migrations/000257_add_stop_resume_status_grace_config.up.sql b/migrations/000257_add_stop_resume_status_grace_config.up.sql new file mode 100644 index 00000000..efa7bceb --- /dev/null +++ b/migrations/000257_add_stop_resume_status_grace_config.up.sql @@ -0,0 +1,7 @@ +-- 停复机成功后的状态保护窗口(add-stop-resume-command-queue 测试中发现的问题修复)。 +-- 背景:停复机成功后上游要几十秒到几分钟才返回新状态,期间查卡状态拿到的是操作前的旧值并被写回库里; +-- 暂停轮询的卡在此期间套餐到期或退款时,判断以为卡已是目标状态而漏停。 +-- polling.stop_resume.status_grace_seconds:停复机成功后多少秒内,查卡状态返回操作前的状态视为上游滞后、不回写;0 表示不保护。 +INSERT INTO tb_system_config (config_key, config_value, value_type, module, description, is_readonly, is_sensitive, creator, updater, created_at, updated_at) VALUES + ('polling.stop_resume.status_grace_seconds', '300', 'int', 'polling', '停复机成功后的状态保护窗口(秒):窗口内查卡状态返回操作前的状态视为上游滞后,不回写;0 不保护', FALSE, FALSE, 0, 0, NOW(), NOW()) +ON CONFLICT (config_key) DO NOTHING; diff --git a/pkg/constants/stop_resume_queue.go b/pkg/constants/stop_resume_queue.go index 67917c78..ff0ec3cf 100644 --- a/pkg/constants/stop_resume_queue.go +++ b/pkg/constants/stop_resume_queue.go @@ -84,6 +84,8 @@ const ( SystemConfigStopResumeUnknownAlert = "polling.stop_resume.unknown_alert_threshold" SystemConfigStopResumeBackoff = "polling.stop_resume.backoff_seconds" SystemConfigStopResumeAlertFailCount = "polling.stop_resume.alert_fail_count" + // SystemConfigStopResumeStatusGrace 停复机成功后的状态保护窗口(秒):窗口内查到操作前的状态视为上游滞后,不回写。0 表示不保护。 + SystemConfigStopResumeStatusGrace = "polling.stop_resume.status_grace_seconds" ) // ========================================