Files
junhong_cmp_fiber/openspec/specs/polling-config-manager/spec.md
huang 434a8b0349
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 9m46s
feat: 轮询系统重构(分片队列 + 停复机统一 + Handler 拆分)
【核心变更】

1. 停复机逻辑统一(StopResumeService)
   - 新增 EvaluateAndAct 统一入口,封装三条件停复机判断
   - 停机条件:无套餐(no_package) / 流量耗尽(traffic_exhausted) / 未实名(not_realname)
   - 复机条件:stop_reason 合规 + 有套餐且未耗尽 + 已实名或行业卡
   - 修复设备套餐 Bug:hasValidPackage 按 device_id 查套餐,而非仅 iot_card_id
   - 设备维度停复机加幂等锁(Redis SetNX,TTL 30s),防止多卡并发重复调 Gateway

2. Redis 分片队列(PollingQueueManager)
   - 新建 queue_manager.go,封装所有轮询 Redis 操作
   - 16 分片 Sorted Set,Key 格式:polling:shard:{shardID}:queue:{taskType}
   - Lua 脚本原子出队(ZRANGEBYSCORE + 分批 ZREM),消除竞态窗口
   - 新增背压检测:队列深度超 50 万时 Scheduler 跳过该分片
   - RemoveFromAllQueues 覆盖 4 种任务类型(含 protect)

3. Handler 拆分(polling_handler.go 1360行 → 5个专注文件)
   - polling_base.go:共享基类(并发控制/卡缓存/重入队)
   - polling_realname_handler.go:实名采集,实名 0→1 时立即触发复机
   - polling_carddata_handler.go:流量采集,保留跨月边界检测逻辑
   - polling_package_handler.go:套餐采集,委托 EvaluateAndAct 决策
   - polling_protect_handler.go:保护期一致性检查,保护期内强制修正

4. 配置管理(PollingConfigManager)
   - 新建 config_manager.go,从 scheduler.go 提取配置职责
   - 内存缓存 + 5 分钟定时刷新,刷新失败保留原缓存
   - 修复 getCardCondition:停机卡返回 suspended,不再错配 activated 配置

5. 渐进式初始化(CardInitializer)
   - 新建 initializer.go,分批加载(每批 10 万),批次间 sleep 500ms
   - 过滤 enable_polling=false 的卡,初始化完成前 Scheduler 不出队

6. 卡生命周期服务(PollingLifecycleService)
   - 新建 lifecycle_service.go,替代已删除的 callbacks.go 和 api_callback.go
   - OnCardCreated/OnCardEnabled/OnCardStatusChanged 入队前检查 enable_polling

7. Scheduler 精简(1000+行 → 227行)
   - 保留纯调度循环:scheduleLoop + processShardSchedule + enqueueBatch
   - 保留每 10 秒触发套餐过期检测和流量重置
   - 移除所有 DB 操作、配置加载、卡初始化逻辑

8. 轮询管控 API(enable_polling)
   - 新增 PUT /api/admin/assets/:id/polling-status 接口
   - 支持对设备/卡维度开关轮询,关闭后从所有分片队列移除

9. 数据库迁移
   - 000103:tb_device 新增 enable_polling 字段(boolean, NOT NULL, DEFAULT true)
   - 000104:新增 suspended 轮询配置,为 activated 配置补全 protect_check_interval

【文件统计】
- 新增:19 个文件(handler × 5、polling 组件 × 4、迁移 × 3 等)
- 修改:20 个文件(bootstrap 注入、store 接口、monitoring 适配分片等)
- 删除:3 个文件(polling_handler.go、callbacks.go、api_callback.go)

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
2026-04-07 12:27:04 +08:00

6.8 KiB
Raw Blame History

Requirement: 独立配置管理模块PollingConfigManager

新建 internal/polling/config_manager.go,提供 PollingConfigManager 类型,从 scheduler.go 中拆分配置相关职责。

文件目标< 150 行职责DB 加载 → 内存缓存 → Redis 同步 → 定时刷新。

从 Scheduler 提取的方法

  • loadConfigs(ctx):从 DB 加载所有启用的 tb_polling_config
  • syncConfigsToRedis(configs):写入 Redis HashTTL 24小时
  • MatchConfig(card):按优先级顺序返回第一个匹配的配置
  • matchConfigConditions(config, card):判断卡是否满足配置条件
  • getCardCondition(card)获取卡的匹配条件carrier、simtype 等)

Scheduler 修改后:不包含任何 DB 查询或配置加载逻辑,通过 configManager.MatchConfig(card) 获取配置。

Scenario: 启动时加载配置并同步 Redis

  • GIVEN Worker 进程启动DB 中有 5 条启用的 PollingConfig
  • WHEN PollingConfigManager.Load(ctx) 被调用
  • THEN 从 DB 加载 5 条配置写入内存缓存sync.RWMutex 保护),同步到 Redis HashTTL 24小时加载完成日志记录「已加载 5 条轮询配置」

Scenario: 配置按优先级匹配

  • GIVEN 内存中有 3 条配置,优先级分别为 1、5、10数字越小优先级越高
  • WHEN 调用 MatchConfig(card) 匹配某张卡
  • THEN 按优先级升序遍历,返回第一个满足所有条件的配置;无匹配时返回 nil该卡不加入轮询队列

Scenario: 配置匹配使用读锁不阻塞

  • GIVEN Scheduler 高频调用 MatchConfig(card)(每秒可能数千次)
  • WHEN 同时有定时刷新正在写入新配置(写锁)
  • THEN 读取操作等待写锁释放后继续,不发生数据竞争;写锁持有时间极短(内存赋值),不影响 Scheduler 调度性能

Requirement: 定时刷新配置5分钟周期

PollingConfigManager.Start(ctx) 启动后台 goroutine每 5 分钟自动重新加载配置。

Scenario: 定时刷新不重启 Worker

  • GIVEN Worker 进程已运行 10 分钟,管理员在控制台修改了某条 PollingConfig 的轮询间隔
  • WHEN 距上次加载超过 5 分钟,定时器触发
  • THEN 自动重新从 DB 加载配置,更新内存缓存,新配置在下一轮调度中生效;无需重启 Worker 进程

Scenario: 刷新失败不影响当前配置

  • GIVEN 定时刷新时 DB 连接超时
  • WHEN Load(ctx) 返回 error
  • THEN 内存缓存保持原有配置不变(不清空),记录 Error 日志「配置刷新失败: xxx」下一轮5分钟后重试

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=trueScheduler 开始正常出队

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 操作
  • 配置加载逻辑(loadConfigssyncConfigsToRedis
  • 卡初始化逻辑(progressiveInitinitCardsBatch
  • 任何 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.Finddb.WhereloadConfig 等字样;文件中出现 HandlePackageActivationCheckHandleDataReset