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>
275 lines
13 KiB
Markdown
275 lines
13 KiB
Markdown
## 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）。
|