From 5d9be1d7e422bf5e1cb2506f3e32815de10d287f Mon Sep 17 00:00:00 2001 From: huang Date: Thu, 16 Apr 2026 14:27:47 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E6=AD=A3=E8=BD=AE=E8=AF=A2?= =?UTF-8?q?=E9=85=8D=E7=BD=AE=E5=A4=9A=E5=8C=B9=E9=85=8D=E9=80=BB=E8=BE=91?= =?UTF-8?q?=EF=BC=8C=E6=94=AF=E6=8C=81=E5=90=8C=E5=8D=A1=E5=8C=B9=E9=85=8D?= =?UTF-8?q?=E5=A4=9A=E4=B8=AA=E9=85=8D=E7=BD=AE=E5=B9=B6=E6=8C=89=20priori?= =?UTF-8?q?ty=20=E5=90=88=E5=B9=B6=20interval?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心变更： - MatchConfig 改为 MatchConfigs，返回所有匹配配置 - MergedTaskIntervals 按 task type 合并各配置，选取最高优先级（非 nil 且最小 priority 值） - hasAnyEnabledInterval 过滤所有 interval 均为 NULL 的配置 - calcInitialDelay 重构为纯函数，接收 interval 参数 - 移除 getEnabledTaskTypes 和 getIntervalByTaskType（被 MergedTaskIntervals 替代） - scheduler.go 新增心跳 key + 顶层 panic recovery + Init 完成守卫 - initializer.go 批量失败日志升级为 Error，逐条检查 Pipeline 命令错误 - 数据迁移：禁用 id=29 的轮询配置（所有 interval 均为 NULL） Co-Authored-By: Claude Opus 4.6 --- cmd/worker/main.go | 1 + internal/polling/config_manager.go | 75 ++++- internal/polling/initializer.go | 70 +++-- internal/polling/lifecycle_service.go | 56 +--- internal/polling/scheduler.go | 28 ++ internal/task/polling_base.go | 14 +- internal/task/polling_utils.go | 30 -- ...00125_disable_polling_config_id29.down.sql | 5 + .../000125_disable_polling_config_id29.up.sql | 7 + .../.openspec.yaml | 2 + .../design.md | 274 ++++++++++++++++++ .../proposal.md | 56 ++++ .../tasks.md | 243 ++++++++++++++++ openspec/specs/polling-config-manager/spec.md | 133 ++------- pkg/constants/redis.go | 7 + 15 files changed, 770 insertions(+), 231 deletions(-) create mode 100644 migrations/000125_disable_polling_config_id29.down.sql create mode 100644 migrations/000125_disable_polling_config_id29.up.sql create mode 100644 openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/.openspec.yaml create mode 100644 openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/design.md create mode 100644 openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/proposal.md create mode 100644 openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/tasks.md diff --git a/cmd/worker/main.go b/cmd/worker/main.go index da226b4..7ddd189 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -171,6 +171,7 @@ func main() { appLogger, ) scheduler := polling.NewScheduler(redisClient, asynqClient, pollingQueueMgr, pollingConfigMgr, appLogger, activationHandler, dataResetHandler) + scheduler.SetInitializer(pollingInitializer) if err := scheduler.Start(ctx); err != nil { appLogger.Error("启动轮询调度器失败", zap.Error(err)) diff --git a/internal/polling/config_manager.go b/internal/polling/config_manager.go index 8e8d601..13c4130 100644 --- a/internal/polling/config_manager.go +++ b/internal/polling/config_manager.go @@ -27,6 +27,12 @@ type PollingConfigManager struct { refreshOnce sync.Once } +// TaskTypeInterval 某任务类型的轮询间隔信息 +type TaskTypeInterval struct { + Interval int // 秒 + Priority int // 来源配置的 priority +} + // NewPollingConfigManager 创建配置管理器 func NewPollingConfigManager(configStore *postgres.PollingConfigStore, redisClient *redis.Client, logger *zap.Logger) *PollingConfigManager { return &PollingConfigManager{ @@ -81,16 +87,79 @@ func (m *PollingConfigManager) Start(ctx context.Context) { // MatchConfig 按优先级匹配第一个符合条件的轮询配置 // 配置按 priority ASC 排序（DB 层保证），数字越小优先级越高 +// MatchConfigs()[0] 的前向兼容包装 func (m *PollingConfigManager) MatchConfig(card *model.IotCard) *model.PollingConfig { + configs := m.MatchConfigs(card) + if len(configs) == 0 { + return nil + } + return configs[0] +} + +// MatchConfigs 按 priority ASC 遍历所有启用的配置，返回所有满足匹配条件且至少有一个非 NULL interval 的配置 +// 无匹配时返回空切片 +func (m *PollingConfigManager) MatchConfigs(card *model.IotCard) []*model.PollingConfig { m.mu.RLock() defer m.mu.RUnlock() + var result []*model.PollingConfig for _, cfg := range m.configs { - if matchConfigConditions(cfg, card) { - return cfg + if matchConfigConditions(cfg, card) && hasAnyEnabledInterval(cfg) { + result = append(result, cfg) } } - return nil + return result +} + +// hasAnyEnabledInterval 检查配置是否有至少一个非 NULL 且大于 0 的轮询间隔 +func hasAnyEnabledInterval(cfg *model.PollingConfig) bool { + return (cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0) || + (cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0) || + (cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0) || + (cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0) || + (cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0) +} + +// MergedTaskIntervals 对所有匹配配置按 priority 遍历，对每种 task type 选取最高优先级的非 nil interval +// 返回 map: taskType → TaskTypeInterval{Interval, Priority} +func (m *PollingConfigManager) MergedTaskIntervals(card *model.IotCard) map[string]TaskTypeInterval { + configs := m.MatchConfigs(card) + merged := make(map[string]TaskTypeInterval) + + for _, cfg := range configs { + priority := int(cfg.Priority) + if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 { + taskType := constants.TaskTypePollingRealname + if _, exists := merged[taskType]; !exists { + merged[taskType] = TaskTypeInterval{Interval: *cfg.RealnameCheckInterval, Priority: priority} + } + } + if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 { + taskType := constants.TaskTypePollingCarddata + if _, exists := merged[taskType]; !exists { + merged[taskType] = TaskTypeInterval{Interval: *cfg.CarddataCheckInterval, Priority: priority} + } + } + if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 { + taskType := constants.TaskTypePollingPackage + if _, exists := merged[taskType]; !exists { + merged[taskType] = TaskTypeInterval{Interval: *cfg.PackageCheckInterval, Priority: priority} + } + } + if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { + taskType := constants.TaskTypePollingProtect + if _, exists := merged[taskType]; !exists { + merged[taskType] = TaskTypeInterval{Interval: *cfg.ProtectCheckInterval, Priority: priority} + } + } + if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { + taskType := constants.TaskTypePollingCardStatus + if _, exists := merged[taskType]; !exists { + merged[taskType] = TaskTypeInterval{Interval: *cfg.CardStatusCheckInterval, Priority: priority} + } + } + } + return merged } // matchConfigConditions 检查卡是否满足配置的匹配条件 diff --git a/internal/polling/initializer.go b/internal/polling/initializer.go index 061f6ed..ac29ed9 100644 --- a/internal/polling/initializer.go +++ b/internal/polling/initializer.go @@ -157,7 +157,8 @@ func (p *PollingInitializer) run(ctx context.Context) { } if initErr := p.initBatch(ctx, cards); initErr != nil { - p.logger.Warn("批量初始化失败", zap.Error(initErr)) + p.logger.Error("批量初始化失败，该批次卡未入轮询队列", + zap.Uint("batch_start_id", lastID), zap.Error(initErr)) } lastID = cards[len(cards)-1].ID @@ -206,16 +207,24 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa if cmdCount == 0 { return } - if _, execErr := pipe.Exec(ctx); execErr != nil { - p.logger.Warn("Pipeline flush 失败，继续下一批", zap.Error(execErr)) + cmds, execErr := pipe.Exec(ctx) + if execErr != nil { + p.logger.Error("Pipeline flush 失败，部分卡可能未入队", zap.Error(execErr)) + } + // 逐条检查，记录具体失败命令（不中断批次） + for _, cmd := range cmds { + if cmdErr := cmd.Err(); cmdErr != nil && cmdErr != redis.Nil { + p.logger.Warn("Pipeline 单条命令失败", + zap.String("cmd", cmd.Name()), zap.Error(cmdErr)) + } } pipe = p.redis.Pipeline() cmdCount = 0 } for _, card := range cards { - cfg := p.configMgr.MatchConfig(card) - if cfg == nil { + intervals := p.configMgr.MergedTaskIntervals(card) + if len(intervals) == 0 { skippedCards++ continue } @@ -224,37 +233,10 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa shardID := int(card.ID) % p.queueMgr.shardCount cardIDStr := fmt.Sprintf("%d", card.ID) - if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 { - nextCheck := calculateNextCheckTime(card.LastRealNameCheckAt, *cfg.RealnameCheckInterval, now) - pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingRealname), redis.Z{ - Score: float64(nextCheck.Unix()), Member: cardIDStr, - }) - cmdCount++ - } - if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 { - nextCheck := calculateNextCheckTime(card.LastDataCheckAt, *cfg.CarddataCheckInterval, now) - pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingCarddata), redis.Z{ - Score: float64(nextCheck.Unix()), Member: cardIDStr, - }) - cmdCount++ - } - if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 { - nextCheck := calculateNextCheckTime(card.LastDataCheckAt, *cfg.PackageCheckInterval, now) - pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingPackage), redis.Z{ - Score: float64(nextCheck.Unix()), Member: cardIDStr, - }) - cmdCount++ - } - if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { - nextCheck := calculateNextCheckTime(card.LastProtectCheckAt, *cfg.ProtectCheckInterval, now) - pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingProtect), redis.Z{ - Score: float64(nextCheck.Unix()), Member: cardIDStr, - }) - cmdCount++ - } - if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { - nextCheck := calculateNextCheckTime(card.LastCardStatusCheckAt, *cfg.CardStatusCheckInterval, now) - pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingCardStatus), redis.Z{ + for taskType, info := range intervals { + lastCheckAt := lastCheckAtByTaskType(card, taskType) + nextCheck := calculateNextCheckTime(lastCheckAt, info.Interval, now) + pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, taskType), redis.Z{ Score: float64(nextCheck.Unix()), Member: cardIDStr, }) cmdCount++ @@ -294,6 +276,22 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa return nil } +// lastCheckAtByTaskType 根据 task type 返回对应的上次检查时间字段 +func lastCheckAtByTaskType(card *model.IotCard, taskType string) *time.Time { + switch taskType { + case constants.TaskTypePollingRealname: + return card.LastRealNameCheckAt + case constants.TaskTypePollingCarddata, constants.TaskTypePollingPackage: + return card.LastDataCheckAt + case constants.TaskTypePollingProtect: + return card.LastProtectCheckAt + case constants.TaskTypePollingCardStatus: + return card.LastCardStatusCheckAt + default: + return nil + } +} + // calculateNextCheckTime 计算下次检查时间 func calculateNextCheckTime(lastCheckAt *time.Time, intervalSeconds int, now time.Time) time.Time { if lastCheckAt == nil { diff --git a/internal/polling/lifecycle_service.go b/internal/polling/lifecycle_service.go index 40bcd54..fbdb417 100644 --- a/internal/polling/lifecycle_service.go +++ b/internal/polling/lifecycle_service.go @@ -9,7 +9,6 @@ import ( "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/internal/store/postgres" - "github.com/break/junhong_cmp_fiber/pkg/constants" ) // PollingLifecycleService 卡生命周期轮询管理服务 @@ -134,68 +133,23 @@ func (s *PollingLifecycleService) shouldEnqueue(ctx context.Context, card *model // enqueueCard 按匹配配置将卡入队分片队列 func (s *PollingLifecycleService) enqueueCard(ctx context.Context, card *model.IotCard) { - cfg := s.configMgr.MatchConfig(card) - if cfg == nil { + intervals := s.configMgr.MergedTaskIntervals(card) + if len(intervals) == 0 { return } - taskTypes := getEnabledTaskTypes(cfg) - for _, taskType := range taskTypes { - if err := s.queueMgr.Requeue(ctx, card.ID, taskType, calcInitialDelay(card, cfg, taskType)); err != nil { + for taskType, info := range intervals { + if err := s.queueMgr.Requeue(ctx, card.ID, taskType, calcInitialDelay(info.Interval)); err != nil { s.logger.Warn("卡入队失败", zap.Uint("card_id", card.ID), zap.String("task_type", taskType), zap.Error(err)) } } } -// getEnabledTaskTypes 返回配置中启用的任务类型列表 -func getEnabledTaskTypes(cfg *model.PollingConfig) []string { - var types []string - if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 { - types = append(types, constants.TaskTypePollingRealname) - } - if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 { - types = append(types, constants.TaskTypePollingCarddata) - } - if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 { - types = append(types, constants.TaskTypePollingPackage) - } - if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { - types = append(types, constants.TaskTypePollingProtect) - } - if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { - types = append(types, constants.TaskTypePollingCardStatus) - } - return types -} - // calcInitialDelay 计算新入队卡的初始检查时间（加随机抖动避免惊群） // 抖动范围 [0, max(interval/10, 2)) 秒；下界取 2 保证 rand.Intn 有实际随机性 // （rand.Intn(1) 永远返回 0，对 interval≤10 的配置会导致零抖动） -func calcInitialDelay(_ *model.IotCard, cfg *model.PollingConfig, taskType string) time.Time { +func calcInitialDelay(interval int) time.Time { now := time.Now() - var interval int - switch taskType { - case constants.TaskTypePollingRealname: - if cfg.RealnameCheckInterval != nil { - interval = *cfg.RealnameCheckInterval - } - case constants.TaskTypePollingCarddata: - if cfg.CarddataCheckInterval != nil { - interval = *cfg.CarddataCheckInterval - } - case constants.TaskTypePollingPackage: - if cfg.PackageCheckInterval != nil { - interval = *cfg.PackageCheckInterval - } - case constants.TaskTypePollingProtect: - if cfg.ProtectCheckInterval != nil { - interval = *cfg.ProtectCheckInterval - } - case constants.TaskTypePollingCardStatus: - if cfg.CardStatusCheckInterval != nil { - interval = *cfg.CardStatusCheckInterval - } - } if interval <= 0 { return now } diff --git a/internal/polling/scheduler.go b/internal/polling/scheduler.go index 992e5a9..81c6d3c 100644 --- a/internal/polling/scheduler.go +++ b/internal/polling/scheduler.go @@ -28,6 +28,7 @@ type Scheduler struct { packageActivationHandler *PackageActivationHandler dataResetHandler *DataResetHandler + initializer *PollingInitializer // 可选，nil 表示不守卫 Init 完成 stopChan chan struct{} wg sync.WaitGroup @@ -103,9 +104,29 @@ func (s *Scheduler) SetStopResumeCallback(callback packagepkg.StopResumeCallback } } +// SetInitializer 注入初始化器（可选，在 Start 前调用） +// Init 未完成时调度器跳过分片出队，消除启动期无效轮询噪音 +func (s *Scheduler) SetInitializer(init *PollingInitializer) { + s.initializer = init +} + +// writeHeartbeat 写入调度器心跳，表明调度器存活 +func (s *Scheduler) writeHeartbeat(ctx context.Context) { + if err := s.redis.Set(ctx, constants.RedisPollingSchedulerHeartbeatKey(), + time.Now().Unix(), 2*s.cfg.ScheduleInterval).Err(); err != nil { + s.logger.Warn("写入调度器心跳失败", zap.Error(err)) + } +} + // scheduleLoop 调度循环 func (s *Scheduler) scheduleLoop(ctx context.Context) { defer s.wg.Done() + defer func() { + if r := recover(); r != nil { + s.logger.Error("调度主循环发生 panic，调度已停止，需重启 Worker", + zap.Any("panic", r), zap.Stack("stack")) + } + }() ticker := time.NewTicker(s.cfg.ScheduleInterval) activationTicker := time.NewTicker(10 * time.Second) @@ -123,6 +144,7 @@ func (s *Scheduler) scheduleLoop(ctx context.Context) { s.logger.Info("调度循环收到 ctx 取消信号") return case <-ticker.C: + s.writeHeartbeat(ctx) s.processShardSchedule(ctx) case <-activationTicker.C: s.processActivationTasks(ctx) @@ -133,10 +155,16 @@ func (s *Scheduler) scheduleLoop(ctx context.Context) { // processShardSchedule 处理手动队列和分片定时队列（每 1 秒触发） // 使用 90% 的 tick 间隔作为超时，确保单分片 Redis 挂起时不阻塞下一个 tick func (s *Scheduler) processShardSchedule(ctx context.Context) { + // 手动队列不受 Init 影响（ConfigManager 已就绪即可） for _, taskType := range allTaskTypes { s.processManualQueue(ctx, taskType, s.cfg.MaxManualBatchSize) } + // Init 未完成时跳过分片扫描，避免空轮询噪音 + if s.initializer != nil && !s.initializer.IsCompleted() { + return + } + if s.queueMgr == nil { return } diff --git a/internal/task/polling_base.go b/internal/task/polling_base.go index 243948f..7a5eaf9 100644 --- a/internal/task/polling_base.go +++ b/internal/task/polling_base.go @@ -102,20 +102,18 @@ func (b *PollingBase) requeueCard(ctx context.Context, cardID uint, taskType str return b.queueMgr.Requeue(ctx, cardID, taskType, time.Now().Add(30*time.Second)) } - cfg := b.configMgr.MatchConfig(card) - if cfg == nil { - // 兜底：配置未加载（DB 暂时不可达）时延迟 30 秒重入队，防止卡从轮询永久消失 + intervals := b.configMgr.MergedTaskIntervals(card) + if len(intervals) == 0 { + // 兜底：配置未加载时延迟 30 秒重入队，防止卡从轮询永久消失 b.logger.Warn("无匹配轮询配置，30秒后重入队（配置可能未加载）", zap.Uint("card_id", cardID), zap.String("task_type", taskType)) return b.queueMgr.Requeue(ctx, cardID, taskType, time.Now().Add(30*time.Second)) } - - interval := getIntervalByTaskType(cfg, taskType) - if interval <= 0 { + info, ok := intervals[taskType] + if !ok || info.Interval <= 0 { b.logger.Debug("轮询间隔为 NULL，不入队", zap.Uint("card_id", cardID), zap.String("task_type", taskType)) return nil } - - nextCheckAt := time.Now().Add(time.Duration(interval) * time.Second) + nextCheckAt := time.Now().Add(time.Duration(info.Interval) * time.Second) return b.queueMgr.Requeue(ctx, cardID, taskType, nextCheckAt) } diff --git a/internal/task/polling_utils.go b/internal/task/polling_utils.go index 1b1d27c..3420775 100644 --- a/internal/task/polling_utils.go +++ b/internal/task/polling_utils.go @@ -7,38 +7,8 @@ import ( "github.com/bytedance/sonic" "go.uber.org/zap" "gorm.io/gorm" - - "github.com/break/junhong_cmp_fiber/internal/model" - "github.com/break/junhong_cmp_fiber/pkg/constants" ) -// getIntervalByTaskType 从配置中提取指定任务类型的轮询间隔（秒） -func getIntervalByTaskType(cfg *model.PollingConfig, taskType string) int { - switch taskType { - case constants.TaskTypePollingRealname: - if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 { - return *cfg.RealnameCheckInterval - } - case constants.TaskTypePollingCarddata: - if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 { - return *cfg.CarddataCheckInterval - } - case constants.TaskTypePollingPackage: - if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 { - return *cfg.PackageCheckInterval - } - case constants.TaskTypePollingProtect: - if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { - return *cfg.ProtectCheckInterval - } - case constants.TaskTypePollingCardStatus: - if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { - return *cfg.CardStatusCheckInterval - } - } - return 0 -} - // shortTaskType 从完整任务类型中提取简短名称（如 polling:carddata → carddata） func shortTaskType(fullTaskType string) string { for i := len(fullTaskType) - 1; i >= 0; i-- { diff --git a/migrations/000125_disable_polling_config_id29.down.sql b/migrations/000125_disable_polling_config_id29.down.sql new file mode 100644 index 0000000..9265d91 --- /dev/null +++ b/migrations/000125_disable_polling_config_id29.down.sql @@ -0,0 +1,5 @@ +-- 回滚：重新启用 id=29 的轮询配置 +UPDATE tb_polling_config +SET status = 1 +WHERE id = 29 + AND status = 0; diff --git a/migrations/000125_disable_polling_config_id29.up.sql b/migrations/000125_disable_polling_config_id29.up.sql new file mode 100644 index 0000000..5f89e5b --- /dev/null +++ b/migrations/000125_disable_polling_config_id29.up.sql @@ -0,0 +1,7 @@ +-- 禁用 id=29 的轮询配置（所有 interval 均为 NULL，导致轮询系统失效） +-- 该配置 priority=1 会匹配所有卡，但其所有 interval 字段全为 NULL， +-- 在多匹配语义下会"吞噬"更具体配置的有效 interval，导致轮询系统失效 +UPDATE tb_polling_config +SET status = 0 +WHERE id = 29 + AND status = 1; diff --git a/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/.openspec.yaml b/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/.openspec.yaml new file mode 100644 index 0000000..859e7bb --- /dev/null +++ b/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-04-15 diff --git a/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/design.md b/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/design.md new file mode 100644 index 0000000..e306398 --- /dev/null +++ b/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/design.md @@ -0,0 +1,274 @@ +## Context + +当前 `PollingConfigManager.MatchConfig(card)` 使用独占式单匹配：遍历所有配置（按 priority ASC），**第一个满足条件的配置直接返回**，后续配置全部忽略。 + +这个设计在以下场景会出问题： +- id=29 (priority=1, card_condition="", 所有 interval 全为 NULL)：匹配所有卡，什么轮询都没配，像黑洞一样废掉整个轮询系统 +- id=28 (priority=25, card_condition="suspended", card_status_check_interval=600)：永远不会有机会被评估，因为被 id=29 截断 + +现有 `polling-config-manager/spec.md` 描述了"按优先级返回第一个匹配"，这是**需求规格说明**，而非实现细节 bug。 + +## Goals / Non-Goals + +**Goals:** +- 支持一张卡匹配多个配置，每个配置贡献自己非 NULL 的 interval +- 相同 task type 从多个匹配的 config 中，按 priority 数值小（优先级高）选取 +- 跳过所有 interval 均为 NULL 的配置（这些配置不应参与轮询匹配） +- 向后兼容：外部已有调用 `MatchConfig` 的地方不受影响 + +**Non-Goals:** +- 不改变配置优先级机制（priority ASC 不变） +- 不改变 DB schema +- 不引入新的外部依赖 + +## Decisions + +### Decision 1: `MatchConfig` → `MatchConfigs`，返回全部匹配配置 + +**选择**：新增 `MatchConfigs(card) []*model.PollingConfig`，返回所有满足条件的配置（已过滤 NULL-interval，已按 priority ASC 排序），保留原 `MatchConfig` 作为其特例（前向兼容）。 + +**替代方案 A（直接改 `MatchConfig` 返回合并结果）**：改动最小，但破坏了 `MatchConfig` 的语义（原 spec 描述为"返回第一个匹配"），且外部若有调用方依赖此行为会无声失效。 + +**替代方案 B（新增 `MatchConfigs` + 废弃 `MatchConfig`）**：API 更干净，但需要迁移所有调用方，成本高。 + +**结论**：采用替代方案 C——新增 `MatchConfigs`，`MatchConfig` 保持不变（返回 `MatchConfigs()[0]`），后续调用方逐步迁移。 + +--- + +### Decision 2: 过滤所有 interval 均为 NULL 的配置 + +**选择**：在 `matchConfigConditions` 之后，新增 `hasAnyEnabledInterval(cfg)` 检查，跳过所有 interval 字段全为 NULL 的配置。 + +**理由**：id=29 这种"所有 interval 全为 NULL"的配置在业务上等效于"不参与轮询"，但当前 `matchConfigConditions` 只检查 card 条件，不检查 interval 是否有效。增加这个过滤可以在配置层面彻底堵住"黑洞"问题。 + +```go +// hasAnyEnabledInterval 检查配置是否有至少一个非 NULL 的轮询间隔 +func hasAnyEnabledInterval(cfg *model.PollingConfig) bool { + return (cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0) || + (cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0) || + (cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0) || + (cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0) || + (cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0) +} +``` + +--- + +### Decision 3: 多配置 interval 合并策略 + +**选择**：新增 `MergedTaskIntervals(card) map[string]int` 方法（在 `config_manager.go`），对所有匹配配置按 priority 遍历，对每种 task type 选取第一个非 NULL 的 interval。 + +**合并逻辑**： +``` +所有匹配 configs（按 priority ASC）: + for each config: + for each taskType in allTaskTypes: + if merged[taskType] == nil && config[taskType] != nil: + merged[taskType] = config[taskType] +``` + +**理由**：`initBatch` 和 `enqueueCard` 都需要这个合并逻辑，集中在 `PollingConfigManager` 中实现可以保持两处调用的一致性。 + +**新增类型**： +```go +// TaskTypeInterval 某任务类型的轮询间隔信息 +type TaskTypeInterval struct { + Interval int // 秒 + Priority int // 来源配置的 priority +} +``` + +--- + +### Decision 4: `initBatch` 中的 interval 选取更新 + +**当前**：`initBatch` 调用 `cfg := p.configMgr.MatchConfig(card)`，然后分别读取 `cfg.CardStatusCheckInterval`、`cfg.PackageCheckInterval` 等字段。 + +**修改后**： +```go +intervals := p.configMgr.MergedTaskIntervals(card) +for taskType, info := range intervals { + nextCheck := calculateNextCheckTime(/* 对应 last 字段 */, info.Interval, now) + // ZADD 到对应分片队列 +} +``` + +每种 task type 对应哪个 `Last*CheckAt` 字段： +| taskType | card 字段 | +|----------|----------| +| realname | LastRealNameCheckAt | +| carddata | LastDataCheckAt | +| package | LastDataCheckAt | +| protect | LastProtectCheckAt | +| card_status | LastCardStatusCheckAt | + +--- + +### Decision 5: `enqueueCard` 中的 interval 选取更新 + +**当前**：`getEnabledTaskTypes(cfg)` 返回 `[]string`，`calcInitialDelay` 分别处理每种 task type。 + +**修改后**：复用 `MergedTaskIntervals`，逻辑与 `initBatch` 一致。`calcInitialDelay` 签名改为 `calcInitialDelay(interval int) time.Time`，接收合并后的 interval 值，内联 jitter 计算。 + +`getEnabledTaskTypes` 移除——`MergedTaskIntervals` 的 key set 天然就是启用的 task type 列表。 + +--- + +### Decision 6: `requeueCard`（`internal/task/polling_base.go`）迁移 + +**发现**：`MatchConfig` 有第三个调用方 `PollingBase.requeueCard()`（`internal/task/polling_base.go:105`），位于 `internal/task/` 包而非 `internal/polling/`。这是**稳态路径**——每次轮询任务完成后的重入队操作，被 5 个 handler 共 26 处调用，是执行频率最高的匹配路径。 + +**当前**： +```go +cfg := b.configMgr.MatchConfig(card) +interval := getIntervalByTaskType(cfg, taskType) +``` + +**修改后**： +```go +intervals := b.configMgr.MergedTaskIntervals(card) +info, ok := intervals[taskType] +if !ok || info.Interval <= 0 { + return nil // 该 task type 无有效配置，不入队 +} +nextCheckAt := time.Now().Add(time.Duration(info.Interval) * time.Second) +return b.queueMgr.Requeue(ctx, cardID, taskType, nextCheckAt) +``` + +**`getIntervalByTaskType` 移除**：`MergedTaskIntervals` 已提供按 task type 查 interval 的能力。`requeueCard` 是 `getIntervalByTaskType` 的唯一调用方，迁移后该函数不再需要。 + +**注意**：`requeueCard` 只需要单个 taskType 的 interval，而 `MergedTaskIntervals` 返回全量 map。这在功能上正确，性能开销可忽略（配置通常 ≤10 条，task type 仅 5 种）。若后续需要优化，可新增 `MergedIntervalForTaskType(card, taskType)` 方法按需停止遍历。 + +--- + +--- + +### Decision 7: `scheduleLoop` 顶层 panic recovery + 心跳机制 + +**问题**：`scheduleLoop` 的主 for-select 循环没有 `recover()`。`processOneShard` 里的子 goroutine 有 recover，但 `processManualQueue`、`processActivationTasks` 等直接在主循环调用，panic 会导致整个调度 goroutine 终止——所有分片队列积压，无任何告警。 + +**修改**： + +```go +func (s *Scheduler) scheduleLoop(ctx context.Context) { + defer s.wg.Done() + defer func() { + if r := recover(); r != nil { + s.logger.Error("调度主循环发生 panic，调度已停止，需重启 Worker", + zap.Any("panic", r)) + } + }() + // ... 原有逻辑 +} +``` + +**心跳机制**：每次 tick 向 Redis 写入心跳 key（`polling:scheduler:heartbeat`），TTL 设为 2 倍 ScheduleInterval（2 秒）。告警规则通过检查该 key 是否存在判断调度器存活。 + +```go +case <-ticker.C: + // 写心跳，TTL=2s（2×1s tick），超过 2s 无更新说明调度器挂了 + _ = s.redis.Set(ctx, constants.RedisPollingSchedulerHeartbeatKey(), + time.Now().Unix(), 2*s.cfg.ScheduleInterval) + s.processShardSchedule(ctx) +``` + +**新增常量**：`pkg/constants/redis.go` 新增 `RedisPollingSchedulerHeartbeatKey() string`，返回 `"polling:scheduler:heartbeat"`。 + +**不新增 AlertRule**：心跳 key 是字符串 key（`GET` 检查 TTL），当前 AlertService 的告警指标是队列深度/成功率，告警规则的具体配置由运维通过 API 创建，不在代码中硬编码。 + +--- + +### Decision 8: Scheduler 注入 `initializer`，Init 完成前守卫 + +**问题**：Scheduler 和 Initializer 同时启动，Init 需要数十分钟加载全量卡。这期间 Scheduler 每秒 tick 出队到空队列，产生无意义的 Redis 读请求和日志噪音。更严重的是，Init 期间手动触发队列的卡会被正常出队，但 `MatchConfig` 此时已可用（ConfigManager 先加载），所以手动触发是正常的——**只需跳过定时出队的空分片扫描**。 + +**修改**： + +`Scheduler` struct 新增可选 `initializer *PollingInitializer` 字段，通过 `SetInitializer(init *PollingInitializer)` 在 `Start` 前注入： + +```go +func (s *Scheduler) processShardSchedule(ctx context.Context) { + // 仍然处理手动队列（允许启动期手动触发） + for _, taskType := range allTaskTypes { + s.processManualQueue(ctx, taskType, s.cfg.MaxManualBatchSize) + } + + // Init 未完成时跳过分片扫描 + if s.initializer != nil && !s.initializer.IsCompleted() { + return + } + // ... 原有分片出队逻辑 +} +``` + +**为何不在 `Start` 前 wait**：同步等待 Init 完成会延迟 Worker 启动（Init 可能需要 30 分钟），不可接受。Skip 方案既保留了手动触发能力，又消除了空扫描噪音。 + +--- + +### Decision 9: `initBatch` 失败日志级别 Warn→Error + +**问题**：`initBatch` 失败时当前记录 Warn 并继续。但 initBatch 失败意味着该批次的卡**未进入任何轮询队列**，除非下次重启重新 Init，这些卡永久丢失调度，是应该触发告警的严重问题。 + +**修改**： + +```go +if initErr := p.initBatch(ctx, cards); initErr != nil { + // Warn → Error：initBatch 失败 = 该批卡未进入轮询队列 + p.logger.Error("批量初始化失败，该批次卡未入队，需关注", + zap.Int("batch_start_id", int(lastID)), zap.Error(initErr)) +} +``` + +**不做**：不重试（批次失败通常是 Redis 连接问题，重试可能级联）。重启 Worker 会触发完整 Init 重跑。 + +--- + +### Decision 10: Pipeline 逐条错误检查 + +**问题**：`initBatch` 中 `flushPipe()` 调用 `pipe.Exec(ctx)` 只检查整体 error。go-redis 的 Pipeline 在部分命令失败时，整体 error 可能是 `MultiError`，也可能是 `nil`（取决于 go-redis 版本和 Redis 模式）。不逐条检查 `cmd.Err()`，会导致部分 ZADD 失败静默丢失。 + +**修改**： + +```go +flushPipe := func() { + if cmdCount == 0 { + return + } + cmds, execErr := pipe.Exec(ctx) + if execErr != nil { + p.logger.Error("Pipeline flush 失败", zap.Error(execErr)) + } + // 逐条检查，记录具体失败的命令 + for _, cmd := range cmds { + if cmd.Err() != nil && cmd.Err() != redis.Nil { + p.logger.Warn("Pipeline 单条命令失败", + zap.String("cmd", cmd.Name()), zap.Error(cmd.Err())) + } + } + pipe = p.redis.Pipeline() + cmdCount = 0 +} +``` + +**注意**：逐条检查仅记录日志，不中断批次。ZADD 失败率极低（仅在 Redis 数据类型冲突等异常情况），单条失败不应阻塞整个批次。 + +--- + +## Risks / Trade-offs + +**[Risk] `MatchConfig` 的前向兼容**：外部若有代码调用 `MatchConfig` 并期望"独占式匹配"，改为返回第一个后可能产生语义变化。 +→ **Mitigation**：`MatchConfig` 行为不变，只是内部调用 `MatchConfigs()[0]`，前向兼容。 + +**[Risk] 临时修复 id=29 的副作用**：用户提到 id=29 当前 `status=1`，所有 interval 全为 NULL。修复后 `MatchConfigs` 会跳过它（因为 `hasAnyEnabledInterval` 返回 false），相当于它从轮询系统"退出"。这是预期行为，但可能用户原本希望它作为 catch-all 兜底配置。 +→ **Mitigation**：已在 design 中明确说明，临时方案是用户手动 `UPDATE tb_polling_config SET status=0 WHERE id=29`。 + +**[Risk] 相同 task type 不同 interval 的优先级覆盖**：极端场景下，配置 A (priority=5, card_status=300) 和配置 B (priority=10, card_status=600) 同时匹配一张卡，按设计会选 A 的 300 秒。但用户可能期望更特定的配置（如 carrier_id 匹配的）优先级高于泛化配置（如 card_condition）。 +→ **当前设计已满足**：priority ASC = 优先级高优先。配置 A 如果 priority 更小（更高优先级），它应该更具体才合理。如果出现"更具体的配置 priority 反而更大"的情况，那是配置管理员的问题，不是代码问题。 + +**[Trade-off] 临时修复 vs 根本修复**：本 design 是根本修复，但需要代码改动 + 部署。临时修复（禁用 id=29）可以立即生效，用户可自行选择。 + +## Open Questions（已闭环） + +1. **~~是否有其他调用方依赖 `MatchConfig` 的独占语义？~~** ✅ **已确认：有。** `internal/task/polling_base.go:105` 的 `requeueCard()` 是第三个调用方，位于 `internal/task/` 包。这是稳态路径（每次轮询任务完成后执行），被 5 个 handler 共 26 处调用。已在 Decision 6 中覆盖迁移方案。 +2. **~~id=29 是否应该被删除而不是禁用？~~** ✅ **决定：添加数据迁移 SQL 禁用。** id=29 作为历史配置，禁用（`status=0`）而非删除——保留审计可追溯性。在 tasks.md 中添加迁移任务。 +3. **~~`calcInitialDelay` 需要改造~~** ✅ **已决定：修改签名为 `calcInitialDelay(interval int) time.Time`。** 详见 Decision 5。同时移除 `getIntervalByTaskType`（详见 Decision 6）。 diff --git a/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/proposal.md b/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/proposal.md new file mode 100644 index 0000000..d8f2031 --- /dev/null +++ b/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/proposal.md @@ -0,0 +1,56 @@ +## Why + +本次变更包含两个独立但相关的修复，均涉及轮询系统的核心可靠性： + +**问题一（配置多匹配）**：当前 `PollingConfigManager.MatchConfig(card)` 的实现是"第一个匹配的配置直接返回，后续配置全部忽略"。这个设计在配置优先级体系下存在致命缺陷：**高 priority 的 catch-all 配置（如 id=29，priority=1，所有 interval 全为 NULL）会匹配所有卡，直接跳过更具体的配置（如 id=28，priority=25，card_status_check_interval=600），导致整个轮询系统失效**。 + +**问题二（调度器稳定性）**：轮询调度器（`Scheduler`）的主循环 `scheduleLoop` 没有顶层 panic recovery 和心跳上报机制。一旦 `processActivationTasks` 等路径发生 panic，整个调度 goroutine 静默终止——所有分片队列持续积压，但 Worker 进程看起来正常，无任何告警。同时，初始化完成前调度器无谓地空轮询；`initBatch` 失败时仅记录 Warn 掩盖了卡初始化缺失的严重性；Pipeline 执行只检查整体 error 而不逐条检查 cmder，部分卡初始化失败会静默丢失。 + +本次修复将匹配逻辑从"独占式单匹配"改为"组合式多匹配"，并同步修复上述调度器稳定性问题。 + +## What Changes + +### 配置多匹配修复 + +- **修改 `MatchConfig` → `MatchConfigs`**：返回所有满足条件的配置（按 priority ASC 排序），而非第一个 +- **修改 `matchConfigConditions`**：跳过所有 interval 均为 NULL 的配置（这些配置不应参与轮询匹配） +- **修改 `initBatch`**：对每种 task type，从所有匹配配置中选取最高优先级的非 NULL interval +- **修改 `enqueueCard`**（`lifecycle_service.go`）：同上，合并多配置的 interval +- **修改 `getEnabledTaskTypes` → 移除**：不再需要，`MergedTaskIntervals` 已包含启用的 task type 列表 +- **重构 `calcInitialDelay`**：修改函数签名接受 `interval int` 参数，移除对 `*model.PollingConfig` 的依赖（因为新逻辑中 interval 来源是 `MergedTaskIntervals` 合并结果，不再属于单一 config） +- **修改 `requeueCard`**（`internal/task/polling_base.go`）：同上，使用 `MergedTaskIntervals` 替代 `MatchConfig` + `getIntervalByTaskType` +- **移除 `getIntervalByTaskType`**（`internal/task/polling_utils.go`）：`MergedTaskIntervals` 已提供按 task type 查 interval 的能力，该辅助函数不再需要 +- **更新 `polling-config-manager` spec**：反映新的多匹配语义 + +### 调度器稳定性修复 + +- **新增 `scheduleLoop` 顶层 panic recovery**：主调度 goroutine 崩溃时记录 Error 日志，防止调度静默停止 +- **新增 Scheduler 心跳机制**：每次 tick 写入 `polling:scheduler:heartbeat`（TTL=2×ScheduleInterval），供告警规则检测调度器存活 +- **新增 Init 完成守卫**：`processShardSchedule` 首先检查 `initializer.IsCompleted()`，Init 未完成时跳过出队，消除启动期无效轮询噪音 +- **`initBatch` 失败日志 Warn→Error**：批量初始化失败意味着部分卡未进入轮询队列，应使用 Error 级别 +- **Pipeline 错误逐条检查**：`pipe.Exec()` 后遍历 `[]redis.Cmder` 逐条检查 `cmd.Err()`，记录具体失败的 ZADD 命令，防止部分卡初始化失败静默丢失 +- **新增 `RedisPollingSchedulerHeartbeatKey()`**（`pkg/constants/redis.go`）：心跳 Key 统一管理 + +## Capabilities + +### Modified Capabilities + +- `polling-config-manager`：原有的"第一个匹配独占"语义改为"所有匹配配置 + 按 task type 选取最高优先级 interval"。这是**需求级别变更**（不仅是实现细节），需要更新 spec 中的 MatchConfig 描述和 Scenario。 +- `polling-scheduler`：新增心跳上报、Init 完成守卫、顶层 panic recovery，调度器可观测性和稳定性提升。 + +## Impact + +- **受影响文件**： + - `internal/polling/config_manager.go`：新增 `MatchConfigs`、`MergedTaskIntervals`、`hasAnyEnabledInterval` + - `internal/polling/initializer.go`：`initBatch` interval 选取逻辑；日志级别 Warn→Error；Pipeline 逐条错误检查 + - `internal/polling/lifecycle_service.go`：`enqueueCard` interval 选取逻辑，移除 `getEnabledTaskTypes` + - `internal/polling/scheduler.go`：新增顶层 panic recovery、心跳写入、Init 守卫；新增 `initializer` 字段 + - `internal/task/polling_base.go`：`requeueCard` interval 选取逻辑（**稳态路径，每次轮询任务完成后执行**） + - `internal/task/polling_utils.go`：移除 `getIntervalByTaskType`（被 `MergedTaskIntervals` 替代） + - `pkg/constants/redis.go`：新增 `RedisPollingSchedulerHeartbeatKey()` +- **不受影响**：API 接口、DB schema、Asynq 任务提交逻辑（`queue_manager.go`）、handler 层 +- **预期收益**： + 1. 配置系统按预期工作，id=28（suspended + card_status_check_interval=600）能正确对停机卡生效 + 2. 调度器故障可被告警规则（检查心跳 key）及时发现，不再静默失效 + 3. 启动期不再产生无效轮询噪音 + 4. 卡初始化失败可见，不再静默丢失 diff --git a/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/tasks.md b/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/tasks.md new file mode 100644 index 0000000..4c3c5ff --- /dev/null +++ b/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/tasks.md @@ -0,0 +1,243 @@ +## 1. config_manager.go 核心修改 + +- [x] 1.1 新增 `TaskTypeInterval` 结构体（`internal/polling/config_manager.go`） + - 字段：`Interval int`，`Priority int` + - 验证：`lsp_diagnostics` 无 error + +- [x] 1.2 新增 `hasAnyEnabledInterval(cfg *model.PollingConfig) bool` 函数 + - 检查所有 5 个 interval 字段（RealnameCheckInterval、CarddataCheckInterval、PackageCheckInterval、ProtectCheckInterval、CardStatusCheckInterval） + - 任一字段非 nil 且 > 0 则返回 true + - 验证：`go build ./internal/polling/...` 成功 + +- [x] 1.3 新增 `MatchConfigs(card *model.IotCard) []*model.PollingConfig` 方法 + - 遍历所有配置（priority ASC），先 `matchConfigConditions` 再 `hasAnyEnabledInterval` + - 返回所有满足条件的配置切片 + - 验证：`go build` 成功 + +- [x] 1.4 修改 `MatchConfig` 委托给 `MatchConfigs` + - `return s.MatchConfigs(card)[0]`，空切片时返回 nil + - 验证：`go build` 成功 + +- [x] 1.5 新增 `MergedTaskIntervals(card *model.IotCard) map[string]TaskTypeInterval` 方法 + - 调用 `MatchConfigs`，按 priority 遍历，对每种 task type 选取最高优先级（priority 数值最小）的非 nil interval + - taskType → `TaskTypeInterval{Interval: xxx, Priority: yyy}` + - 验证：`go build` 成功 + +## 2. initializer.go 使用新 interval 合并逻辑 + +- [x] 2.1 修改 `initBatch` 中的 interval 选取逻辑 + - 将 `cfg := p.configMgr.MatchConfig(card)` 改为 `intervals := p.configMgr.MergedTaskIntervals(card)` + - 原来逐个 `if cfg.XXXInterval != nil` 判断，改为 `for taskType, info := range intervals` + - 每种 task type 对应的 `Last*CheckAt` 字段映射： + - `realname` → `card.LastRealNameCheckAt` + - `carddata` → `card.LastDataCheckAt` + - `package` → `card.LastDataCheckAt` + - `protect` → `card.LastProtectCheckAt` + - `card_status` → `card.LastCardStatusCheckAt` + - 需要新增 `lastCheckAtByTaskType(card, taskType) *time.Time` 辅助函数集中映射关系，避免 initBatch 内联大量 switch + - 验证：`go build ./internal/polling/...` 成功，`go vet ./internal/polling/...` 无 warning + +## 3. lifecycle_service.go 使用新 interval 合并逻辑 + calcInitialDelay 重构 + +> 原 Task 3.1 和 Task 4.3 描述完全相同的改动，已合并为本节。 + +- [x] 3.1 重构 `calcInitialDelay` 函数签名 + - 当前签名：`calcInitialDelay(_ *model.IotCard, cfg *model.PollingConfig, taskType string) time.Time` + - 新签名：`calcInitialDelay(interval int) time.Time` + - 内联 jitter 计算逻辑：`jitterMax := max(interval/10, 2); return now.Add(rand.Intn(jitterMax) * time.Second)` + - 验证：`go build ./internal/polling/...` 成功 + +- [x] 3.2 修改 `enqueueCard` 使用 `MergedTaskIntervals` + - 将 `cfg := s.configMgr.MatchConfig(card)` + `taskTypes := getEnabledTaskTypes(cfg)` 改为 `intervals := s.configMgr.MergedTaskIntervals(card)` + - 空 map 时直接 return（等效于原 `cfg == nil`） + - 遍历 `intervals`，对每种 taskType 调用 `s.queueMgr.Requeue(ctx, card.ID, taskType, calcInitialDelay(info.Interval))` + - 验证：`go build ./internal/polling/...` 成功 + +- [x] 3.3 移除 `getEnabledTaskTypes` 函数 + - grep 确认仅 `enqueueCard` 调用（已确认无其他调用方） + - `MergedTaskIntervals` 返回的 map key set 天然就是启用的 task type 列表，该函数不再需要 + - 验证：`go build` 成功 + +## 4. polling_base.go 迁移（稳态路径，P0 修复） + +> ⚠️ 这是执行频率最高的匹配路径——每次轮询任务完成后的重入队操作，被 5 个 handler 共 26 处调用。 +> 原提案遗漏此调用方（`internal/task/polling_base.go:105`），位于 `internal/task/` 包而非 `internal/polling/`。 + +- [x] 4.1 修改 `requeueCard` 使用 `MergedTaskIntervals`（`internal/task/polling_base.go`） + - 当前代码： + ```go + cfg := b.configMgr.MatchConfig(card) + if cfg == nil { /* 兜底 30s 重入队 */ } + interval := getIntervalByTaskType(cfg, taskType) + if interval <= 0 { return nil } + ``` + - 改为： + ```go + intervals := b.configMgr.MergedTaskIntervals(card) + if len(intervals) == 0 { + // 兜底：配置未加载时延迟 30 秒重入队，防止卡永久消失 + return b.queueMgr.Requeue(ctx, cardID, taskType, time.Now().Add(30*time.Second)) + } + info, ok := intervals[taskType] + if !ok || info.Interval <= 0 { + // 该 task type 无有效配置，不入队 + return nil + } + nextCheckAt := time.Now().Add(time.Duration(info.Interval) * time.Second) + return b.queueMgr.Requeue(ctx, cardID, taskType, nextCheckAt) + ``` + - 注意保留原有的兜底逻辑（配置未加载时 30 秒重入队） + - 验证：`go build ./internal/task/...` 成功 + +- [x] 4.2 移除 `getIntervalByTaskType`（`internal/task/polling_utils.go`） + - `MergedTaskIntervals` 已提供按 task type 查 interval 的能力 + - grep 确认 `getIntervalByTaskType` 仅 `requeueCard` 调用（已确认唯一调用方） + - 删除 `polling_utils.go` 中的 `getIntervalByTaskType` 函数 + - 验证：`go build ./internal/task/...` 成功 + +## 5. 数据迁移与外部确认 + +- [x] 5.1 确认外部调用方已全部迁移 + - `grep -r "MatchConfig(" --include="*.go" . | grep -v "_test.go"` + - 预期结果：仅 `config_manager.go` 中的定义和前向兼容调用 + - 确认 `getIntervalByTaskType` 和 `getEnabledTaskTypes` 已无调用方 + - 验证：`go build ./...` 成功 + +- [x] 5.2 添加数据迁移 SQL 禁用 id=29 + - 创建迁移文件：`UPDATE tb_polling_config SET status = 0 WHERE id = 29 AND status = 1` + - 添加 down 迁移：`UPDATE tb_polling_config SET status = 1 WHERE id = 29 AND status = 0` + - 验证：迁移文件格式符合 `db-migration` skill 规范 + +## 6. 更新 spec.md 反映多匹配语义 + +- [x] 6.1 更新 `polling-config-manager/spec.md` 的 MatchConfig 描述 + - 说明从"返回第一个匹配"改为"返回所有匹配配置" + - 添加 `MatchConfigs` 和 `MergedTaskIntervals` 的 API 描述 + - 注：spec.md 已提前更新，确认内容与最终实现一致即可 + +## 7. 验证与构建（配置多匹配） + +- [x] 7.1 全量 `go build ./...` 确认编译通过 +- [x] 7.2 `go vet ./internal/polling/... ./internal/task/...` 无 error +- [x] 7.3 `lsp_diagnostics` 对所有修改文件无 error +- [ ] 7.4 手动验证：确认 `MatchConfigs` 对一张 suspended 卡返回 id=28（而非 id=29） +- [ ] 7.5 手动验证：确认 `MergedTaskIntervals` 对一张同时匹配多配置的卡，相同 task type 选 priority 数值小的 +- [ ] 7.6 手动验证：id=29（所有 interval 全为 NULL）被 `hasAnyEnabledInterval` 正确过滤 +- [ ] 7.7 手动验证：`requeueCard` 路径——对停机卡执行 card_status 轮询后，能正确重入队（interval 来自 id=28 而非 id=29） + +## 8. 调度器稳定性修复 + +### 8.1 新增 `RedisPollingSchedulerHeartbeatKey()`（`pkg/constants/redis.go`） + +- [x] 8.1.1 在轮询相关 Key 区块新增函数： + ```go + // RedisPollingSchedulerHeartbeatKey 轮询调度器心跳 Key + // 调度器每次 tick 写入当前时间戳，TTL=2s + // 告警规则：key 不存在超过 5s 视为调度器停止运行 + func RedisPollingSchedulerHeartbeatKey() string { + return "polling:scheduler:heartbeat" + } + ``` + - 验证：`go build ./pkg/constants/...` 成功 + +### 8.2 `scheduler.go` 新增顶层 panic recovery + 心跳 + +- [x] 8.2.1 在 `scheduleLoop` 函数起始处新增顶层 `defer recover()`： + - 紧接在 `defer s.wg.Done()` 之后添加 + - panic 时记录 Error 日志，包含 `zap.Any("panic", r)` 和 `zap.Stack("stack")` 字段 + - 验证：`go build ./internal/polling/...` 成功 + +- [x] 8.2.2 在 `scheduler.go` 中新增 `writeHeartbeat(ctx context.Context)` 方法： + ```go + func (s *Scheduler) writeHeartbeat(ctx context.Context) { + if err := s.redis.Set(ctx, constants.RedisPollingSchedulerHeartbeatKey(), + time.Now().Unix(), 2*s.cfg.ScheduleInterval).Err(); err != nil { + s.logger.Warn("写入调度器心跳失败", zap.Error(err)) + } + } + ``` + - 验证：`go build ./internal/polling/...` 成功 + +- [x] 8.2.3 在 `scheduleLoop` 的 `ticker.C` 分支调用 `writeHeartbeat`： + ```go + case <-ticker.C: + s.writeHeartbeat(ctx) // 心跳 + s.processShardSchedule(ctx) + ``` + - 验证：`go build ./internal/polling/...` 成功 + +### 8.3 `scheduler.go` 新增 Init 完成守卫 + +- [x] 8.3.1 `Scheduler` struct 新增 `initializer *PollingInitializer` 字段（可选，nil 表示不守卫） + +- [x] 8.3.2 新增 `SetInitializer(init *PollingInitializer)` 方法（在 Start 前调用）： + ```go + func (s *Scheduler) SetInitializer(init *PollingInitializer) { + s.initializer = init + } + ``` + +- [x] 8.3.3 修改 `processShardSchedule` 新增 Init 守卫（**只守卫分片出队，不守卫手动队列**）： + ```go + func (s *Scheduler) processShardSchedule(ctx context.Context) { + // 手动队列不受 Init 影响（ConfigManager 已就绪即可） + for _, taskType := range allTaskTypes { + s.processManualQueue(ctx, taskType, s.cfg.MaxManualBatchSize) + } + + // Init 未完成时跳过分片扫描，避免空轮询噪音 + if s.initializer != nil && !s.initializer.IsCompleted() { + return + } + + // ... 原有 queueMgr 判空 + 分片并发出队逻辑 + } + ``` + - 注意：原有 `for _, taskType := range allTaskTypes { s.processManualQueue(...) }` 代码块已在上面，需要移除函数体中重复的部分 + - 验证：`go build ./internal/polling/...` 成功 + +- [x] 8.3.4 在 `cmd/worker/main.go` 中 `scheduler.Start(ctx)` 之前调用 `scheduler.SetInitializer(pollingInitializer)` + - 验证：`go build ./cmd/worker/...` 成功 + +### 8.4 `initializer.go` 日志级别 + Pipeline 逐条检查 + +- [x] 8.4.1 将 `initBatch` 失败的日志级别从 Warn 升级为 Error： + ```go + // 修改前 + p.logger.Warn("批量初始化失败", zap.Error(initErr)) + // 修改后 + p.logger.Error("批量初始化失败，该批次卡未入轮询队列", + zap.Uint("batch_start_id", lastID), zap.Error(initErr)) + ``` + - 验证：`go build ./internal/polling/...` 成功 + +- [x] 8.4.2 修改 `flushPipe` 闭包，新增逐条 `cmd.Err()` 检查： + ```go + flushPipe := func() { + if cmdCount == 0 { + return + } + cmds, execErr := pipe.Exec(ctx) + if execErr != nil { + p.logger.Error("Pipeline flush 失败，部分卡可能未入队", zap.Error(execErr)) + } + // 逐条检查，记录具体失败命令（不中断批次） + for _, cmd := range cmds { + if cmdErr := cmd.Err(); cmdErr != nil && cmdErr != redis.Nil { + p.logger.Warn("Pipeline 单条命令失败", + zap.String("cmd", cmd.Name()), zap.Error(cmdErr)) + } + } + pipe = p.redis.Pipeline() + cmdCount = 0 + } + ``` + - 验证：`go build ./internal/polling/...` 成功 + +## 9. 验证与构建（调度器稳定性） + +- [x] 9.1 全量 `go build ./...` 确认编译通过 +- [x] 9.2 `lsp_diagnostics` 对 `scheduler.go`、`initializer.go`、`pkg/constants/redis.go` 无 error +- [ ] 9.3 手动验证：Worker 启动后，`polling:scheduler:heartbeat` key 存在（TTL≈2s），每秒刷新 +- [ ] 9.4 手动验证：Worker 启动时（Init 进行中），日志中无 `分片出队` 相关记录（只有手动队列处理日志） +- [ ] 9.5 手动验证：Init 完成后，`分片出队` 日志正常出现 diff --git a/openspec/specs/polling-config-manager/spec.md b/openspec/specs/polling-config-manager/spec.md index d5e5036..806c885 100644 --- a/openspec/specs/polling-config-manager/spec.md +++ b/openspec/specs/polling-config-manager/spec.md @@ -1,116 +1,43 @@ -### Requirement: 独立配置管理模块（PollingConfigManager） +## MODIFIED Requirements -新建 `internal/polling/config_manager.go`，提供 `PollingConfigManager` 类型，从 `scheduler.go` 中拆分配置相关职责。 +### Requirement: 按优先级顺序返回所有匹配配置（多匹配 + interval 过滤） -**文件目标**：< 150 行，职责：DB 加载 → 内存缓存 → Redis 同步 → 定时刷新。 +原描述： +> `MatchConfig(card)`：按优先级顺序返回第一个匹配的配置 -**从 Scheduler 提取的方法**： -- `loadConfigs(ctx)`：从 DB 加载所有启用的 `tb_polling_config` -- `syncConfigsToRedis(configs)`：写入 Redis Hash（TTL 24小时） -- `MatchConfig(card)`：按优先级顺序返回第一个匹配的配置 -- `matchConfigConditions(config, card)`：判断卡是否满足配置条件 -- `getCardCondition(card)`：获取卡的匹配条件（carrier、simtype 等） +修改为： -**Scheduler 修改后**：不包含任何 DB 查询或配置加载逻辑，通过 `configManager.MatchConfig(card)` 获取配置。 +`MatchConfigs(card) []*model.PollingConfig` 按 priority ASC 遍历所有启用的配置，返回**所有满足匹配条件且至少有一个非 NULL interval**的配置。无匹配时返回空切片。 -#### Scenario: 启动时加载配置并同步 Redis -- **GIVEN** Worker 进程启动，DB 中有 5 条启用的 PollingConfig -- **WHEN** `PollingConfigManager.Load(ctx)` 被调用 -- **THEN** 从 DB 加载 5 条配置，写入内存缓存（sync.RWMutex 保护），同步到 Redis Hash（TTL 24小时）；加载完成日志记录「已加载 5 条轮询配置」 +`matchConfigConditions(config, card)` 判断卡是否满足配置条件时，**额外过滤**所有 polling interval 均为 NULL 的配置（`RealnameCheckInterval`、`CarddataCheckInterval`、`PackageCheckInterval`、`ProtectCheckInterval`、`CardStatusCheckInterval` 均无有效值）。 -#### Scenario: 配置按优先级匹配 -- **GIVEN** 内存中有 3 条配置，优先级分别为 1、5、10（数字越小优先级越高） -- **WHEN** 调用 `MatchConfig(card)` 匹配某张卡 -- **THEN** 按优先级升序遍历，返回第一个满足所有条件的配置；无匹配时返回 nil，该卡不加入轮询队列 +#### Scenario: 多配置同时匹配一张卡 +- **GIVEN** 配置 A (priority=5, card_condition="", card_status_check_interval=600, protect_check_interval=nil) 和配置 B (priority=10, card_condition="suspended", card_status_check_interval=nil, package_check_interval=60) 均处于启用状态 +- **WHEN** 调用 `MatchConfigs(card)`，其中 card.network_status=0（suspended） +- **THEN** 返回 [配置 A, 配置 B]，按 priority ASC 排序；配置 B 的 `package_check_interval=60` 来自配置 A 不覆盖的 task type -#### Scenario: 配置匹配使用读锁不阻塞 -- **GIVEN** Scheduler 高频调用 `MatchConfig(card)`（每秒可能数千次） -- **WHEN** 同时有定时刷新正在写入新配置（写锁） -- **THEN** 读取操作等待写锁释放后继续，不发生数据竞争；写锁持有时间极短（内存赋值），不影响 Scheduler 调度性能 +#### Scenario: 所有 interval 均为 NULL 的配置被过滤 +- **GIVEN** 配置 C (priority=1, card_condition="", 所有 interval 全为 NULL) 和配置 D (priority=25, card_condition="suspended", card_status_check_interval=600) 均处于启用状态 +- **WHEN** 调用 `MatchConfigs(card)` 对任意卡 +- **THEN** 返回结果**不包含**配置 C（因为没有任何有效 interval）；配置 D 正常返回 ---- +#### Scenario: 无任何配置匹配 +- **GIVEN** 没有任何启用的配置，或所有配置的条件都不满足卡属性 +- **WHEN** 调用 `MatchConfigs(card)` +- **THEN** 返回空切片 `[]` -### Requirement: 定时刷新配置（5分钟周期） +### Requirement: 多配置 Interval 合并选取 -`PollingConfigManager.Start(ctx)` 启动后台 goroutine，每 5 分钟自动重新加载配置。 +新增 `MergedTaskIntervals(card) map[string]TaskTypeInterval` 方法。 -#### Scenario: 定时刷新不重启 Worker -- **GIVEN** Worker 进程已运行 10 分钟，管理员在控制台修改了某条 PollingConfig 的轮询间隔 -- **WHEN** 距上次加载超过 5 分钟，定时器触发 -- **THEN** 自动重新从 DB 加载配置，更新内存缓存，新配置在下一轮调度中生效；无需重启 Worker 进程 +对 `MatchConfigs(card)` 返回的所有配置，按 priority ASC 遍历，对每种 task type（realname、carddata、package、protect、card_status）选取**最高优先级（priority 数值最小）的那个 config 的 interval**。 -#### Scenario: 刷新失败不影响当前配置 -- **GIVEN** 定时刷新时 DB 连接超时 -- **WHEN** `Load(ctx)` 返回 error -- **THEN** 内存缓存保持原有配置不变（不清空），记录 Error 日志「配置刷新失败: xxx」；下一轮（5分钟后）重试 +#### Scenario: 相同 task type 不同 interval 按优先级选取 +- **GIVEN** 配置 A (priority=5, card_status_check_interval=300) 和配置 B (priority=10, card_status_check_interval=600) 同时匹配某张停机卡 +- **WHEN** 调用 `MergedTaskIntervals(card)` +- **THEN** 结果中 `card_status` 的 interval 为 300（来自 priority=5 的配置 A） ---- - -### Requirement: 独立卡初始化模块（CardInitializer） - -新建 `internal/polling/initializer.go`，提供 `CardInitializer` 类型，从 `scheduler.go` 中拆分渐进式初始化职责。 - -**文件目标**：< 250 行，职责：分批 DB 查询 → 分片路由入队 → 进度追踪 → enable_polling 过滤。 - -**从 Scheduler 提取的方法**： -- `progressiveInit(ctx)`：分批加载卡并入队 -- `initCardsBatch(ctx, offset, limit)`：加载单批次卡 -- `initCardPolling(ctx, card)`：为单张卡匹配配置并入队 - -**Scheduler 修改后**：不包含任何初始化逻辑，通过 `cardInitializer.Run(ctx)` 异步启动，通过 `cardInitializer.GetProgress()` 查询进度。 - -#### Scenario: 渐进式分批初始化避免 DB 过载 -- **GIVEN** DB 中有 500 万张启用轮询的卡 -- **WHEN** `CardInitializer.Run(ctx)` 被调用 -- **THEN** 分批加载（每批 10 万张），批次间 Sleep 500ms；约 500 批次，每批处理后记录进度日志「初始化进度: 100000/5000000」；全部完成后 `initCompleted=true`，Scheduler 开始正常出队 - -#### Scenario: 跳过 enable_polling=false 的卡 -- **GIVEN** 10 万张卡中，5000 张卡的 `enable_polling=false` -- **WHEN** 批次处理时遇到这 5000 张卡 -- **THEN** 这 5000 张卡不加入任何轮询队列（ZADD 跳过）；其余 95000 张卡按正常流程入队 - -#### Scenario: 初始化幂等（重启不重复） -- **GIVEN** Worker 重启，部分卡已在分片队列中（上次初始化留下） -- **WHEN** 重新执行 `CardInitializer.Run(ctx)` -- **THEN** ZADD 操作幂等：若卡已在队列中，只更新 score 为新的初始延迟时间（不添加重复条目）；Redis Sorted Set 中不出现重复成员 - -#### Scenario: Scheduler 只在初始化完成后出队 -- **GIVEN** Worker 刚启动，初始化尚未完成（`initCompleted=false`） -- **WHEN** Scheduler 的 `scheduleLoop` 执行 -- **THEN** 检查 `cardInitializer.GetProgress().Completed`，若为 false 则跳过本轮所有分片的出队；避免初始化中途 Scheduler 取到不完整队列 - -#### Scenario: 进度暴露给监控接口 -- **GIVEN** 管理员调用轮询状态监控接口 -- **WHEN** `MonitoringService` 调用 `cardInitializer.GetProgress()` -- **THEN** 返回 `{Total: 5000000, Processed: 1500000, Completed: false}`；前端可展示初始化进度百分比 - ---- - -### Requirement: Scheduler 精简到 < 250 行（Mi1 修复） - -> **注意**：本 spec 原写 `< 200 行`，已与 `design.md` 决策 9 和 `tasks.md` 统一更正为 **`< 250 行`**。 -> 原因：精简后的 Scheduler 保留了套餐过期检测和流量重置两个触发器（见决策 9），额外占用约 50 行。 - -`internal/polling/scheduler.go` 精简后只保留纯调度循环逻辑，所有子职责委托给新建的三个模块。 - -**精简后 Scheduler 的完整职责**： -1. 启动 `CardInitializer.Run(ctx)`（异步） -2. 启动 `PollingConfigManager.Start(ctx)`（定时刷新） -3. 运行 `scheduleLoop`：每秒 × N分片 × 4任务类型 的出队循环 -4. 背压检测：`GetQueueDepth > 阈值` 时跳过该分片 - -**禁止出现在精简后 Scheduler 中的内容**： -- 任何 `db.Find()`、`db.Where()` 等 DB 操作 -- 配置加载逻辑（`loadConfigs`、`syncConfigsToRedis`） -- 卡初始化逻辑（`progressiveInit`、`initCardsBatch`） -- 任何 Callback 注册（`callbacks.go` 已删除） - -#### Scenario: Scheduler 保留套餐过期检测和流量重置触发 -- **GIVEN** Phase 4+5 原子部署完成，精简后的 Scheduler 运行中 -- **WHEN** 每 10 秒定时器触发 -- **THEN** 调用 `PackageActivationHandler.HandlePackageActivationCheck(ctx)` 检查过期套餐；调用 `DataResetHandler.HandleDataReset(ctx)` 检查流量重置；这两个触发器在精简后 Scheduler 中**必须保留**（决策 9） - -#### Scenario: Scheduler 精简后行数验证 -- **GIVEN** Phase 4+5 重构完成 -- **WHEN** 统计 `internal/polling/scheduler.go` 的代码行数 -- **THEN** 文件行数（含注释）**< 250 行**（因保留套餐过期/流量重置触发器，此处统一以 design.md 决策 9 和 tasks.md 的 <250 为准）；文件中不出现 `db.Find`、`db.Where`、`loadConfig` 等字样；文件中出现 `HandlePackageActivationCheck` 和 `HandleDataReset` +#### Scenario: 不同 task type 来自不同配置 +- **GIVEN** 配置 A (priority=5, card_status_check_interval=300, package_check_interval=nil) 和配置 B (priority=10, card_status_check_interval=nil, package_check_interval=60) 同时匹配某张停机卡 +- **WHEN** 调用 `MergedTaskIntervals(card)` +- **THEN** 结果中 `card_status` interval=300（来自配置 A），`package` interval=60（来自配置 B） diff --git a/pkg/constants/redis.go b/pkg/constants/redis.go index 90a9656..a36c670 100644 --- a/pkg/constants/redis.go +++ b/pkg/constants/redis.go @@ -428,6 +428,13 @@ func RedisPollingDeviceOpLockKey(deviceID uint) string { return fmt.Sprintf("polling:device:op_lock:%d", deviceID) } +// RedisPollingSchedulerHeartbeatKey 轮询调度器心跳 Key +// 调度器每次 tick 写入当前时间戳，TTL=2s +// 告警规则：key 不存在超过 5s 视为调度器停止运行 +func RedisPollingSchedulerHeartbeatKey() string { + return "polling:scheduler:heartbeat" +} + // ======================================== // 支付配置缓存 Redis Key // ========================================