junhong_cmp_fiber/openspec/changes/archive/2026-04-16-fix-polling-config-multi-match/tasks.md at 2b3a9cb33fbf88daee9f68a6dcacc4e4c48678e7

csxj2026/junhong_cmp_fiber

Files

构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m27s

Details

fix: 修正轮询配置多匹配逻辑，支持同卡匹配多个配置并按 priority 合并 interval

核心变更：
- 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 <noreply@anthropic.com>

2026-04-16 14:27:47 +08:00

11 KiB

Raw Blame History

1. config_manager.go 核心修改

1.1 新增 TaskTypeInterval 结构体（internal/polling/config_manager.go）
- 字段：Interval int，Priority int
- 验证：lsp_diagnostics 无 error
1.2 新增 hasAnyEnabledInterval(cfg *model.PollingConfig) bool 函数
- 检查所有 5 个 interval 字段（RealnameCheckInterval、CarddataCheckInterval、PackageCheckInterval、ProtectCheckInterval、CardStatusCheckInterval）
- 任一字段非 nil 且 > 0 则返回 true
- 验证：go build ./internal/polling/... 成功
1.3 新增 MatchConfigs(card *model.IotCard) []*model.PollingConfig 方法
- 遍历所有配置（priority ASC），先 matchConfigConditions 再 hasAnyEnabledInterval
- 返回所有满足条件的配置切片
- 验证：go build 成功
1.4 修改 MatchConfig 委托给 MatchConfigs
- return s.MatchConfigs(card)[0]，空切片时返回 nil
- 验证：go build 成功
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 合并逻辑

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 描述完全相同的改动，已合并为本节。

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/... 成功
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/... 成功
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/。

4.1 修改 requeueCard 使用 MergedTaskIntervals（internal/task/polling_base.go）

当前代码：

cfg := b.configMgr.MatchConfig(card)
if cfg == nil { /* 兜底 30s 重入队 */ }
interval := getIntervalByTaskType(cfg, taskType)
if interval <= 0 { return nil }

改为：

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/... 成功

4.2 移除 getIntervalByTaskType（internal/task/polling_utils.go）
- MergedTaskIntervals 已提供按 task type 查 interval 的能力
- grep 确认 getIntervalByTaskType 仅 requeueCard 调用（已确认唯一调用方）
- 删除 polling_utils.go 中的 getIntervalByTaskType 函数
- 验证：go build ./internal/task/... 成功

5. 数据迁移与外部确认

5.1 确认外部调用方已全部迁移
- grep -r "MatchConfig(" --include="*.go" . | grep -v "_test.go"
- 预期结果：仅 config_manager.go 中的定义和前向兼容调用
- 确认 getIntervalByTaskType 和 getEnabledTaskTypes 已无调用方
- 验证：go build ./... 成功
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 反映多匹配语义

6.1 更新 polling-config-manager/spec.md 的 MatchConfig 描述
- 说明从"返回第一个匹配"改为"返回所有匹配配置"
- 添加 MatchConfigs 和 MergedTaskIntervals 的 API 描述
- 注：spec.md 已提前更新，确认内容与最终实现一致即可

7. 验证与构建（配置多匹配）

7.1 全量 go build ./... 确认编译通过
7.2 go vet ./internal/polling/... ./internal/task/... 无 error
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`）

8.1.1 在轮询相关 Key 区块新增函数：

// RedisPollingSchedulerHeartbeatKey 轮询调度器心跳 Key
// 调度器每次 tick 写入当前时间戳，TTL=2s
// 告警规则：key 不存在超过 5s 视为调度器停止运行
func RedisPollingSchedulerHeartbeatKey() string {
    return "polling:scheduler:heartbeat"
}

验证：go build ./pkg/constants/... 成功

8.2 `scheduler.go` 新增顶层 panic recovery + 心跳

8.2.1 在 scheduleLoop 函数起始处新增顶层 defer recover()：
- 紧接在 defer s.wg.Done() 之后添加
- panic 时记录 Error 日志，包含 zap.Any("panic", r) 和 zap.Stack("stack") 字段
- 验证：go build ./internal/polling/... 成功

8.2.2 在 scheduler.go 中新增 writeHeartbeat(ctx context.Context) 方法：

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/... 成功

8.2.3 在 scheduleLoop 的 ticker.C 分支调用 writeHeartbeat：
```
case <-ticker.C:
    s.writeHeartbeat(ctx) // 心跳
    s.processShardSchedule(ctx)
```
- 验证：go build ./internal/polling/... 成功

8.3 `scheduler.go` 新增 Init 完成守卫

8.3.1 Scheduler struct 新增 initializer *PollingInitializer 字段（可选，nil 表示不守卫）

8.3.2 新增 SetInitializer(init *PollingInitializer) 方法（在 Start 前调用）：

func (s *Scheduler) SetInitializer(init *PollingInitializer) {
    s.initializer = init
}

8.3.3 修改 processShardSchedule 新增 Init 守卫（只守卫分片出队，不守卫手动队列）：

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/... 成功

8.3.4 在 cmd/worker/main.go 中 scheduler.Start(ctx) 之前调用 scheduler.SetInitializer(pollingInitializer)
- 验证：go build ./cmd/worker/... 成功

8.4 `initializer.go` 日志级别 + Pipeline 逐条检查

8.4.1 将 initBatch 失败的日志级别从 Warn 升级为 Error：

// 修改前
p.logger.Warn("批量初始化失败", zap.Error(initErr))
// 修改后
p.logger.Error("批量初始化失败，该批次卡未入轮询队列",
    zap.Uint("batch_start_id", lastID), zap.Error(initErr))

验证：go build ./internal/polling/... 成功

8.4.2 修改 flushPipe 闭包，新增逐条 cmd.Err() 检查：

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. 验证与构建（调度器稳定性）

9.1 全量 go build ./... 确认编译通过
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 完成后，分片出队 日志正常出现

11 KiB Raw Blame History Unescape Escape