huang
5d9be1d7e4
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m27s
核心变更： - 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>
244 lines
11 KiB
Markdown
244 lines
11 KiB
Markdown
## 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 完成后，`分片出队` 日志正常出现
|