## 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 完成后,`分片出队` 日志正常出现