csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
Iteration/8-11
junhong_cmp_fiber/openspec/changes/add-polling-async-ingest/tasks.md
邹进 1214181e78
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 16m2s
Details
feat: 轮询观测入库异步化，调网关与入库分离，采集记录经消息攒批写入 
- 轮询步骤异步模式下只调网关并发布 card.ingest 入库消息，入库消费者固定并发 30、按采集 ID 去重、可重试错误不限次数退避重投
- 采集记录经 card.collection 攒批写入，失败整批重投，新增 (observation_id, observed_at) 唯一索引
- 网络、实名观测增加乱序保护；轮询读查询走缓存，套餐状态查询各自计时且失败不缓存
- 入库积压告警与反压；系统配置 polling.ingest.mode 默认 sync 可回退；测试环境 nsqd 消息全部落盘

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-06 16:33:29 +09:00

4.6 KiB
Raw Permalink Blame History

1. 幂等与乱序基础

  • 1.1 采集队列 Lua 按 observed_at 去重（现有 Lua 已按同一采集时间去重，无需改动）
  • 1.2 ApplyNetworkObservation 增加采集时间早于 last_card_status_check_at 时不改状态的保护
  • 1.3 ApplyCardObservation 增加采集时间早于 last_real_name_check_at 时不改状态的保护
  • 1.4 迁移：采集成功表、失败表新增 (observation_id, observed_at) 唯一索引（成对 up/down）

2. 配置

  • 2.1 启动配置 polling.ingest.concurrency（30）、polling.ingest.timeout（30s）、polling.collection.batch_size（200），含默认值、环境变量与校验
  • 2.2 系统配置 polling.ingest.mode（async/sync）、polling.ingest.alert_depth（200000）、polling.ingest.pause_depth（500000，须大于告警阈值）
  • 2.3 启动连接预算检查：入库并发 + 1 > MAX_OPEN_CONNS − 20 时记 ERROR

3. 采集记录批量写

  • 3.1 定义 card.collection 消息与发布器（成功、失败两种记录）
  • 3.2 批量消费者：缓冲按 200 条或 1 秒刷写，多行 INSERT ON CONFLICT DO NOTHING，成功统一 FIN、失败统一重投
  • 3.3 ApplyTrafficObservation 采集记录改为发布消息，发布失败同步写；删除 collectionRecordTimeout 与"写失败只记告警"
  • 3.4 测试环境验证：攒批、整批失败重投、重复记录忽略（项目自动化测试为 N/A，按 ENG-TEST-001 在测试环境验证）

4. 入库消息与消费者

  • 4.1 定义 card.ingest 消息（三种观测、原始数据、优先项、发布时间）与编解码，补测试
  • 4.2 入库消费者：已处理标记检查、按类型调用 Apply*Observation、成功写标记（48h）
  • 4.3 错误分类与重投：可重试错误按 5s/15s/30s/1m/2m/5m 延迟重投、不限次数，第 10 次起记 ERROR；业务性错误记 WARN 后 FIN
  • 4.4 收尾归属：优先执行保持同步入库（优先项完成、对接日志沿用原逻辑）；StopPolling 由步骤按网关结果移出调度；入库段写 flow_evaluated:<card_id>
  • 4.5 在 cmd/worker/carrier_scheduler.go 注册入库与采集记录消费者（MaxAttempts=0），随 worker 优雅退出
  • 4.6 测试环境验证：重复投递、提交后未写标记重放、卡不存在、可重试错误重投

5. 轮询步骤改造

  • 5.1 流量步骤：异步模式下网关成功后发布入库消息即成功返回；网关失败直接发布采集失败记录
  • 5.2 卡状态步骤、实名步骤：异步模式下发布入库消息
  • 5.3 发布失败降级为同步入库并记 WARN；sync 模式走原逻辑
  • 5.4 套餐步骤跳过依据改为读 flow_evaluated:<card_id>
  • 5.5 测试环境验证：异步成功、发布失败降级、同步模式行为不变、套餐步骤跳过

6. 读查询缓存

  • 6.1 步骤读卡改为读卡缓存（去掉每步先清缓存再查库）；入库段提交后失效缓存
  • 6.2 有活动优先项的卡集合：入队加、终态删、启动全量重建；步骤先查集合
  • 6.3 运营商实名开关、重置日进程内缓存 1 分钟；设备轮询开关与绑定关系并入卡缓存
  • 6.4 lazyTierFacts 每次查询单独计时；Prefill 失败的卡不写缓存

7. 积压监控与反压

  • 7.1 leader 每分钟读 nsqd stats，记录 card.ingest、card.collection 积压与最老等待时间
  • 7.2 超过告警阈值或等待 > 10 分钟记 ERROR
  • 7.3 超过反压阈值置暂停标记，分发器暂停领取新卡；低于一半恢复

8. 部署（测试环境）

  • 8.1 docker-compose.prod.yml nsqd 参数改 --mem-queue-size=0
  • 8.2 go build ./...、相关包测试、OpenSpec --strict 校验通过
  • 8.3 提交并推送 Iteration/8-11（自动部署测试环境），polling.ingest.mode 先保持 sync
  • 8.4 测试库执行唯一索引迁移（用户授权后）

9. 测试环境验证

  • 9.1 切 async 后 1 小时：context deadline exceeded 从每小时约 8,000 次降到个位数
  • 9.2 慢查询告警与 worker 日志量明显下降
  • 9.3 采集记录对账：入库消息数 = 成功表 + 失败表新增行数
  • 9.4 抽样卡日快照、扣减与同步模式一致；重投与乱序不重复扣减
  • 9.5 card.ingest 常态等待 < 30 秒
  • 9.6 切回 sync 行为恢复；停 nsqd 时步骤降级同步入库
  • 9.7 nsqd 重启后未处理消息继续消费

10. 归档

  • 10.1 add-carrier-polling-scheduler、add-card-traffic-queue-and-observed-deduction、refactor-audit-log-persistence 归档后再归档本 Change
Reference in New Issue View Git Blame Copy Permalink