junhong_cmp_fiber/openspec/changes/add-polling-async-ingest/design.md

Context

依据 docs/2026-11-06-轮询入库异步化技术方案.md。按运营商账户调度上线后，测试环境 worker 实测同时处理约 380 个任务（理论上限约 1,290：卡轮询 41 × 20、停复机 41 × 10、设备评估 10、异步任务队列 50），共用 MAX_OPEN_CONNS=60 的连接池。每个轮询步骤在同一个任务里先调网关、再同步写库，写库计入步骤超时；结果是每小时约 8,000 次 SQL 超时、13.8 万条慢查询告警、采集记录因 3 秒写超时丢失。

现状（流量步骤）：读优先项 → 清缓存后查卡 → 范围校验查库 → 调网关 → ApplyTrafficObservation（Redis 队列 Push + 事务：锁卡、查套餐、写快照、更新卡、outbox、通道阈值、审计）→ 写采集记录（3s）→ 对接日志、优先项完成 → 返回 TriggeredEvaluation。卡状态、实名步骤结构相同。

约束：基于 Iteration/8-11，只部署测试环境；测试环境网关只读开关保持开启；不改计量与扣减业务规则。

Goals / Non-Goals

Goals:

• 调网关的并发继续按账户控制；入库并发固定、可配置，与账户数无关。
• 入库不丢、不重复算，乱序安全。
• 轮询步骤调网关前的读查询基本不碰数据库。
• 一键回退到同步入库；nsqd 故障时自动降级。

Non-Goals:

• 死信表与重放接口（本期不做）。
• 调整按账户调度参数、网关限流规则。
• 生产环境发布。
• 人工刷新、运营商回调、激活前结算、观测序列的异步化。

Decisions

D1 入库拆到单独的 NSQ topic，而不是给卡任务加全局并发上限

备选：在卡任务消费者上加一个全局信号量。拒绝原因：卡任务里既有网关等待又有写库，全局上限会同时压低网关吞吐；而网关并发本来就由账户限流控制。拆开后网关侧按账户并发，数据库侧固定 30 并发，两种瓶颈各自控制。

D2 入库复用现有 Apply*Observation

入库消费者把消息还原为 domain.TrafficObservation / NetworkObservation / RealnameObservation 后调用现有应用服务，计量、扣减、通道阈值、审计逻辑不复制、不改写。同步路径与异步路径共用一套业务代码，polling.ingest.mode=sync 时轮询步骤直接调用同一函数。

D3 消息格式

card.ingest 一条消息一次观测，JSON，字段见规格。observed_at 取网关返回时刻，入库全程使用它，保证延迟入库不改变日期归属。网关原始数据按现有截断规则（4KB）在发布前截断，单条约 2KB。

D4 重试用 NSQ 延迟重投、不限次数

入库失败按错误分类：可重试错误（数据库超时、连接错误、锁等待超时、context deadline exceeded）调用 msg.RequeueWithoutBackoff(delay)，延迟按 attempts 取 5s/15s/30s/1m/2m/5m…；消费者 MaxAttempts=0（不限）。nsqd --max-msg-timeout=15m 已满足最大 5 分钟延迟。业务性错误（CodeNotFound、校验失败）FIN 并记 WARN。第 10 次起每次重投记 ERROR。

卡任务的业务重试仍由 Redis score 表达（现有要求），入库 topic 与卡任务 topic 互不影响。

D5 幂等

• 处理前查 Redis ingest:done:<observation_id>，存在即 FIN；事务提交后 SET … EX 48h。
• 事务已提交但标记未写入时的重放：日快照取当日最大、增量按"本月最大 − 已结算"得 0；outbox 按 event_id ON CONFLICT DO NOTHING（已核对 internal/infrastructure/messaging/outbox/repository.go:103）；采集记录唯一索引兜底。
• 采集队列 Lua 增加同 observed_at 去重，防止重放时同一点进队两次影响判断。

D6 网络、实名观测的乱序保护

现有 ApplyNetworkObservation、ApplyCardObservation 没有比较采集时间。新增：事务内锁卡后，若 observed_at 早于卡上 last_card_status_check_at / last_real_name_check_at，不改状态、不写事件，只确认消息并记 INFO。同步入口也走同一判断，行为一致。

D7 采集记录走 card.collection 批量写

成功记录在入库事务提交后发布；网关失败、无读数的失败记录由轮询步骤直接发布。批量消费者 max_in_flight=500，内存缓冲按 batch_size=200 或 1 秒刷写，多行 INSERT … ON CONFLICT DO NOTHING，成功统一 FIN、失败统一 Requeue。写库并发 1。唯一索引 (observation_id, observed_at)，observed_at 为分区键，满足分区表唯一索引要求。删除 collectionRecordTimeout 与"写失败只记告警"。

D8 收尾动作的归属

原位置（步骤内）	异步后
优先项完成、对接日志"业务事实发生变化"	本轮认领了优先项的观测保持同步入库，沿用原逻辑；普通轮询本来就不写对接日志
StopPolling → removeCard	步骤按网关结果（ShouldStopPollingForRisk）直接返回停止轮询；入库段写库时关闭卡轮询开关
延迟确认的观测值	步骤按网关状态映射（MapGatewayNetworkStatus）得出
TriggeredEvaluation 跳过套餐步骤	入库段写 Redis flow_evaluated:<card_id>，TTL = 套餐检查间隔；套餐步骤读取

理由：优先项需要"本轮写库后完成"的语义，且量很小（测试环境十几分钟几条），保持同步比在入库段重建优先项轮次简单可靠。停止轮询与延迟确认只依赖网关结果，不需要等入库。

步骤结果在网关成功、消息发布成功时即为成功，写回下次到期时间；入库失败不影响该卡调度退避。

D9 读查询缓存

• 卡信息：步骤不再"先清缓存再查库"，改读 Redis 卡缓存；入库段事务提交后失效缓存（现有 cache.Invalidate）。
• 优先项：Redis 集合 polling:priority:active（成员 card_id:task_type），优先项提示下发时加入；leader 每 30 秒按库内活动项重建（并入 2 分钟内新入队的项），就绪标记存在且不在集合中时不查库，否则回退查库。
• 运营商实名开关（已有）、重置日：进程内缓存 1 分钟，重置日读取失败不缓存。
• 设备轮询开关与绑定关系：卡任务的范围校验结果按卡进程内缓存 30 秒；卡自身轮询开关仍实时判断，生命周期事件仍实时校验。
• lazyTierFacts：3 秒总超时改为每次查询单独 2 秒；Prefill 查询失败的卡不写缓存。

D10 积压监控与反压

worker leader 每分钟请求 nsqd /stats?format=json&topic=…，取 topic 未分发量与 channel 待消费量、延迟重投量之和作为积压，等待时间取 leader 进程内最近处理消息的 now − published_at，写 INFO 指标；超过 alert_depth（默认 200,000）或等待 > 10 分钟写 ERROR。超过 pause_depth（默认 500,000）时在 Redis 置 polling:ingest:paused，分发器每轮检查，置位时不领取新卡；低于一半时清除。

D11 同步回退与降级

polling.ingest.mode 放系统配置（5 秒进程缓存），默认 sync，部署后确认无误再切 async。采集记录不论模式都经 card.collection 写入（nsqd 不可用时同步写）。sync 时步骤直接调用 Apply*Observation 并同步写采集记录（采集记录仍可发 card.collection；nsqd 不可用时同步写）。异步模式发布失败时，该次观测降级为同步入库并记 WARN。

D12 连接预算检查

启动时：ingest.concurrency + 1（采集记录） ≤ MAX_OPEN_CONNS − 20，否则 ERROR（不阻止启动）。测试环境 30 + 1 ≤ 40。

D13 nsqd 落盘

测试环境 docker-compose.prod.yml nsqd 参数改 --mem-queue-size=0，所有 topic 写 ./nsq-data。每秒几十条消息，磁盘写入可忽略；积压 50 万条约 1GB。

Risks / Trade-offs

• [入库延迟使后台数据晚几秒] → 常态目标等待 < 30 秒；积压告警覆盖异常。
• [停复机、通道阈值判断晚几秒] → 判断逻辑不变，只是触发时间后移；与网关调用间隔相比可接受。
• [不做死信，坏消息无限重试] → 业务性错误直接丢弃；可重试错误第 10 次起 ERROR 告警，积压告警兜底；需要时后续加死信。
• [乱序保护改变同步入口行为] → 只拦"比已应用观测更旧"的状态变化，同步入口原本也不应接受更旧的观测。
• [卡缓存陈旧导致轮询范围判断滞后] → 入库段与停复机、轮询开关修改均失效缓存；缓存 TTL 保持现有值。
• [nsqd 全落盘性能] → 当前量级每秒几十条，远低于 nsqd 磁盘队列能力。
• [一张卡的写入跨两个消费者，排查变复杂] → 日志统一带 observation_id、card_id。

Migration Plan

1. 部署含新代码的镜像，polling.ingest.mode 默认 sync，确认行为不变（采集记录已改经 card.collection 写入）。
2. 执行采集记录唯一索引迁移（测试库，用户授权后执行）。
3. nsqd 改 --mem-queue-size=0 并重启 nsqd。
4. 系统配置切 async，观察 1 小时：SQL 超时数、慢查询数、积压、采集记录对账。
5. 回退：系统配置切 sync，已在队列中的消息继续被消费完。

Open Questions

• 无。2026-11-06 已定：入库并发 30；nsqd --mem-queue-size=0；不做死信；告警 200,000 / 反压 500,000。