fix: 关闭轮询的卡不再重新入队
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m32s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m32s
关闭轮询时会移出队列并清除卡缓存,但执行收尾重入队、卡状态检查补齐套餐任务、限流重入队 都不判断轮询开关,执行中途被关闭的卡会被放回队列并互相补齐,关闭轮询无法生效。 - 卡缓存写入 enable_polling,旧缓存缺字段时从库补读并回写 - 重入队、补齐缺失任务、限流重入队前判断轮询开关,已关闭则不入队
This commit is contained in:
@@ -87,6 +87,12 @@ func (b *PollingBase) requeueRateLimited(cardID uint, taskType string) error {
|
||||
delay = configured
|
||||
}
|
||||
}
|
||||
ctx, cancel := pollingFallbackContext()
|
||||
card, err := b.getCardWithCache(ctx, cardID)
|
||||
cancel()
|
||||
if err == nil && !card.EnablePolling {
|
||||
return nil
|
||||
}
|
||||
return b.requeueCardAt(cardID, taskType, time.Now().Add(delay))
|
||||
}
|
||||
|
||||
@@ -189,6 +195,10 @@ func (b *PollingBase) ensureMissingTask(ctx context.Context, cardID uint, taskTy
|
||||
return err
|
||||
}
|
||||
|
||||
if !card.EnablePolling {
|
||||
// 已关闭轮询的卡不补齐任务,否则卡状态与套餐任务会互相补回队列,关闭轮询无法生效。
|
||||
return nil
|
||||
}
|
||||
info, ok := b.configMgr.MergedTaskIntervals(card)[taskType]
|
||||
if !ok || info.Interval <= 0 {
|
||||
return nil
|
||||
@@ -231,6 +241,11 @@ func (b *PollingBase) pollingNextCheckAt(ctx context.Context, cardID uint, taskT
|
||||
zap.Uint("card_id", cardID), zap.String("task_type", taskType), zap.Error(err))
|
||||
return time.Now().Add(pollingRequeueFallbackInterval), true
|
||||
}
|
||||
if !card.EnablePolling {
|
||||
// 执行中途被关闭轮询的卡不再重入队(关闭时已移出队列并清除缓存,此处读到的是库中最新值)。
|
||||
b.logger.Info("卡已关闭轮询,不再重入队", zap.Uint("card_id", cardID), zap.String("task_type", taskType))
|
||||
return time.Time{}, false
|
||||
}
|
||||
intervals := b.configMgr.MergedTaskIntervals(card)
|
||||
if len(intervals) == 0 {
|
||||
// 兜底:配置未加载时延迟 30 秒重入队,防止卡从轮询永久消失
|
||||
|
||||
@@ -18,8 +18,10 @@ func (b *PollingBase) getCardWithCache(ctx context.Context, cardID uint) (*model
|
||||
result, err := b.redis.HGetAll(ctx, key).Result()
|
||||
if err == nil && len(result) > 0 {
|
||||
card := parseCardFromCache(cardID, result)
|
||||
if _, ok := result["asset_status"]; !ok {
|
||||
b.backfillAssetStatus(ctx, key, card)
|
||||
_, hasAssetStatus := result["asset_status"]
|
||||
_, hasEnablePolling := result["enable_polling"]
|
||||
if !hasAssetStatus || !hasEnablePolling {
|
||||
b.backfillCardFields(ctx, key, card)
|
||||
}
|
||||
return card, nil
|
||||
}
|
||||
@@ -36,16 +38,20 @@ func (b *PollingBase) getCardWithCache(ctx context.Context, cardID uint) (*model
|
||||
return card, nil
|
||||
}
|
||||
|
||||
// backfillAssetStatus 为分层轮询补齐旧缓存缺少的资产状态:只读取并回写这一个字段,
|
||||
// 不整条重建缓存,避免覆盖缓存中单独维护的读数等字段。
|
||||
func (b *PollingBase) backfillAssetStatus(ctx context.Context, key string, card *model.IotCard) {
|
||||
var statuses []int
|
||||
if err := b.db.WithContext(ctx).Model(&model.IotCard{}).Where("id = ?", card.ID).Limit(1).Pluck("asset_status", &statuses).Error; err != nil || len(statuses) == 0 {
|
||||
// backfillCardFields 为旧缓存补齐资产状态与轮询开关:只读取并回写这两个字段,
|
||||
// 不整条重建缓存,避免覆盖缓存中单独维护的读数等字段。读取失败时保留缓存解析结果(轮询开关按开启)。
|
||||
func (b *PollingBase) backfillCardFields(ctx context.Context, key string, card *model.IotCard) {
|
||||
var rows []struct {
|
||||
AssetStatus int
|
||||
EnablePolling bool
|
||||
}
|
||||
if err := b.db.WithContext(ctx).Model(&model.IotCard{}).Select("asset_status, enable_polling").
|
||||
Where("id = ?", card.ID).Limit(1).Scan(&rows).Error; err != nil || len(rows) == 0 {
|
||||
return
|
||||
}
|
||||
card.AssetStatus = statuses[0]
|
||||
if err := b.redis.HSet(ctx, key, "asset_status", card.AssetStatus).Err(); err != nil {
|
||||
b.logger.Debug("回写卡缓存资产状态失败", zap.Uint("card_id", card.ID), zap.Error(err))
|
||||
card.AssetStatus, card.EnablePolling = rows[0].AssetStatus, rows[0].EnablePolling
|
||||
if err := b.redis.HSet(ctx, key, "asset_status", card.AssetStatus, "enable_polling", boolToStr(card.EnablePolling)).Err(); err != nil {
|
||||
b.logger.Debug("回写卡缓存字段失败", zap.Uint("card_id", card.ID), zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -122,6 +128,11 @@ func parseCardFromCache(cardID uint, result map[string]string) *model.IotCard {
|
||||
card.AssetStatus = status
|
||||
}
|
||||
}
|
||||
// 轮询开关:所有关闭/开启路径都会清除卡缓存,缓存中的取值与库一致;旧缓存缺字段时按开启并由读取方回填。
|
||||
card.EnablePolling = true
|
||||
if v, ok := result["enable_polling"]; ok {
|
||||
card.EnablePolling = v == "1" || v == "true"
|
||||
}
|
||||
if v, ok := result["is_standalone"]; ok {
|
||||
card.IsStandalone = v == "1" || v == "true"
|
||||
} else {
|
||||
@@ -158,6 +169,7 @@ func writeCardToCache(b *PollingBase, key string, card *model.IotCard) {
|
||||
"data_usage_mb": card.DataUsageMB,
|
||||
"stop_reason": card.StopReason,
|
||||
"is_standalone": boolToStr(card.IsStandalone),
|
||||
"enable_polling": boolToStr(card.EnablePolling),
|
||||
"asset_status": card.AssetStatus,
|
||||
"cached_at": time.Now().Unix(),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user