diff --git a/openspec/changes/add-carrier-polling-scheduler/tasks.md b/openspec/changes/add-carrier-polling-scheduler/tasks.md index 90f9cd30..f1d3674e 100644 --- a/openspec/changes/add-carrier-polling-scheduler/tasks.md +++ b/openspec/changes/add-carrier-polling-scheduler/tasks.md @@ -1,75 +1,75 @@ ## 0. 前置确认 -- [ ] 0.1 确认第 1 个 Change(限流器)与第 3 个 Change(去掉保护期)已上线;维护者提供测试环境 nsqd + nsqlookupd 地址。 +- [x] 0.1 确认第 1 个 Change(限流器)与第 3 个 Change(去掉保护期)已上线;维护者提供测试环境 nsqd + nsqlookupd 地址。(第 1、3 个 Change 已部署测试环境;测试环境 NSQ 由 docker-compose.prod.yml 新增 nsqlookupd + nsqd 提供,Worker 注入 nsqd:4150 / nsqlookupd:4161) ## 1. 迁移与配置 -- [ ] 1.1 新增成对迁移:`tb_carrier_polling_param`(`carrier_id` 唯一、`inflight_limit`、`card_max_in_flight`、`stop_resume_max_in_flight`、`sr_inflight_limit`、`priority_max_ratio`、`stop_resume_min_interval`、`stop_resume_confirm_delay`、审计字段)。 -- [ ] 1.2 新增成对迁移:`tb_polling_check_scene`(`scene` 唯一、`check_types`、`confirm_delays`、`cooldown`),预置 8 个场景,读入口场景带默认冷却。 -- [ ] 1.3 新增成对迁移:`tb_polling_failure_alert`(卡、运营商账户、检查类型、连续失败、首次与最近失败、最近错误、状态、处理人与时间;部分唯一索引 `(card_id, check_type) WHERE status = 未处理`)。 -- [ ] 1.4 新增成对迁移:`tb_system_config` 插入 `polling.scheduler.mode`(默认 `legacy`)、步骤顺序、单步超时、卡任务总超时余量、退避序列、告警阈值、账户级告警窗口与阈值;`tb_polling_concurrency_config` 表注释标记废弃。 -- [ ] 1.5 `pkg/config` 与 `defaults/config.yaml`:NSQ 地址、`msg_timeout`、租约、续期间隔、分发间隔、优雅退出宽限期、默认执行上限与 `max_in_flight`;`go.mod` 引入 NSQ 客户端。 -- [ ] 1.6 注册受控系统配置项并实现 JSON 结构校验(步骤类型合法且不重复、退避序列递增)。 +- [x] 1.1 新增成对迁移:`tb_carrier_polling_param`(`carrier_id` 唯一、`inflight_limit`、`card_max_in_flight`、`stop_resume_max_in_flight`、`sr_inflight_limit`、`priority_max_ratio`、`stop_resume_min_interval`、`stop_resume_confirm_delay`、审计字段)。(000251) +- [x] 1.2 新增成对迁移:`tb_polling_check_scene`(`scene` 唯一、`check_types`、`confirm_delays`、`cooldown`),预置 8 个场景,读入口场景带默认冷却。(000252,预置 8 个场景) +- [x] 1.3 新增成对迁移:`tb_polling_failure_alert`(卡、运营商账户、检查类型、连续失败、首次与最近失败、最近错误、状态、处理人与时间;部分唯一索引 `(card_id, check_type) WHERE status = 未处理`)。(000253;部分唯一索引含告警级别与账户,账户级告警 card_id=0) +- [x] 1.4 新增成对迁移:`tb_system_config` 插入 `polling.scheduler.mode`(默认 `legacy`)、步骤顺序、单步超时、卡任务总超时余量、退避序列、告警阈值、账户级告警窗口与阈值;`tb_polling_concurrency_config` 表注释标记废弃。(000254,默认 legacy) +- [x] 1.5 `pkg/config` 与 `defaults/config.yaml`:NSQ 地址、`msg_timeout`、租约、续期间隔、分发间隔、优雅退出宽限期、默认执行上限与 `max_in_flight`;`go.mod` 引入 NSQ 客户端。(polling.carrier.*;go-nsq v1.1.0) +- [x] 1.6 注册受控系统配置项并实现 JSON 结构校验(步骤类型合法且不重复、退避序列递增)。(carrierqueue.RegisterConfigDefinitions,后台系统配置可编辑) ## 2. Redis 结构与 Lua -- [ ] 2.1 常量:账户队列、running、dispatch_lock,卡 due、claim、exec 的 key。 -- [ ] 2.2 领取 Lua:清理过期 running、执行上限、优先占比、预计调用数对比本秒剩余 QPS、`SET claim NX PX`、推后 score、写 running;已有 claim 时推后到剩余 TTL 后。 -- [ ] 2.3 续期 Lua:校验 token,延长 claim、exec,推后 score 与 running。 -- [ ] 2.4 收尾 Lua:校验 token,删 claim、exec,移出 running,写 due 与失败计数,按优先标记设置 score;token 不一致时只写业务结果。 -- [ ] 2.5 请求检查 Lua:设 due[type]=now、冷却判断、放入 priority 或打优先标记,合并重复请求。 +- [x] 2.1 常量:账户队列、running、dispatch_lock,卡 due、claim、exec 的 key。 +- [x] 2.2 领取 Lua:清理过期 running、执行上限、优先占比、预计调用数对比本秒剩余 QPS、`SET claim NX PX`、推后 score、写 running;已有 claim 时推后到剩余 TTL 后。(按限流组剩余 QPS 在 Go 侧计算后传入 Lua;只统计模式不按 QPS 限制) +- [x] 2.3 续期 Lua:校验 token,延长 claim、exec,推后 score 与 running。 +- [x] 2.4 收尾 Lua:校验 token,删 claim、exec,移出 running,写 due 与失败计数,按优先标记设置 score;token 不一致时只写业务结果。 +- [x] 2.5 请求检查 Lua:设 due[type]=now、冷却判断、放入 priority 或打优先标记,合并重复请求。(另支持普通通道:手动触发只把到期设为现在、不进优先队列) ## 3. 初始化与生命周期 -- [ ] 3.1 `initializer.go`:按批读库计算各类型 due(复用第 2 个 Change 的分层与重置日前到期计算),写 due 与账户 queue;进度写入 init-progress。 -- [ ] 3.2 `lifecycle_service.go`、`asset_polling_service.go`:增删卡、状态变化、启停轮询改为操作账户队列与 due;恢复轮询时 due 设为当前加抖动。 -- [ ] 3.3 优先提示(`priority_manual_service.go`)与手动触发(`manual_trigger_service.go`)改为写 due 与账户队列。 +- [x] 3.1 `initializer.go`:按批读库计算各类型 due(复用第 2 个 Change 的分层与重置日前到期计算),写 due 与账户 queue;进度写入 init-progress。(CarrierInitializer;进度在 Worker 内与 /polling-stats/carrier-queues 的 init_state 可见,未并入旧 init-progress 接口) +- [x] 3.2 `lifecycle_service.go`、`asset_polling_service.go`:增删卡、状态变化、启停轮询改为操作账户队列与 due;恢复轮询时 due 设为当前加抖动。(API 与 Worker 的生命周期服务按模式写入;资产轮询开关经生命周期回调生效) +- [x] 3.3 优先提示(`priority_manual_service.go`)与手动触发(`manual_trigger_service.go`)改为写 due 与账户队列。(旧提示通道与手动触发 List 由新分发器每秒消费后写入账户队列,接口不变) ## 4. 分发器 -- [ ] 4.1 Worker 内每秒对各账户抢分发锁并执行领取 Lua,发布到 `polling.card.{cid}`;发布失败按 token 删 claim、score 恢复为当前时间。 -- [ ] 4.2 读取账户调度参数(缺省用默认),执行上限不超过订阅进程 `max_in_flight` 之和,不满足时收紧并上报。 -- [ ] 4.3 定时动作:统计各账户处理中标记到期次数、写分发器心跳。 +- [x] 4.1 Worker 内每秒对各账户抢分发锁并执行领取 Lua,发布到 `polling.card.{cid}`;发布失败按 token 删 claim、score 恢复为当前时间。 +- [x] 4.2 读取账户调度参数(缺省用默认),执行上限不超过订阅进程 `max_in_flight` 之和,不满足时收紧并上报。 +- [x] 4.3 定时动作:统计各账户处理中标记到期次数、写分发器心跳。 ## 5. NSQ 消费者与卡任务 -- [ ] 5.1 NSQ 生产者与消费者封装:按账户动态订阅与取消订阅、`max_in_flight`、`Touch`、优雅退出(停止分发与订阅、宽限期内完成后 FIN)。 -- [ ] 5.2 消费入口:核对 claim token、抢 exec 执行权,任一失败即 FIN。 -- [ ] 5.3 将实名、流量、卡状态、套餐 4 个 handler 拆为步骤函数(`StepInput` 含可选外部结果);按配置顺序串行执行到期类型;单步超时、总超时、panic 恢复。 -- [ ] 5.4 执行期间按续期间隔续期;续期失败立即停止后续步骤。 -- [ ] 5.5 网关调用使用限流器等待模式;限流等待超时按短延迟重排不计失败。 -- [ ] 5.6 沿用 `polling_priority_claim.go` 的认领与收尾;保留卡流量同步锁。 +- [x] 5.1 NSQ 生产者与消费者封装:按账户动态订阅与取消订阅、`max_in_flight`、`Touch`、优雅退出(停止分发与订阅、宽限期内完成后 FIN)。(internal/infrastructure/nsqbus) +- [x] 5.2 消费入口:核对 claim token、抢 exec 执行权,任一失败即 FIN。 +- [x] 5.3 将实名、流量、卡状态、套餐 4 个 handler 拆为步骤函数(`StepInput` 含可选外部结果);按配置顺序串行执行到期类型;单步超时、总超时、panic 恢复。(四个 handler 拆出 Step,旧 Asynq Handle 只映射重入队;StepInput 外部结果暂未实现,按需在批量查询时扩展) +- [x] 5.4 执行期间按续期间隔续期;续期失败立即停止后续步骤。 +- [x] 5.5 网关调用使用限流器等待模式;限流等待超时按短延迟重排不计失败。(ctx 标记等待模式) +- [x] 5.6 沿用 `polling_priority_claim.go` 的认领与收尾;保留卡流量同步锁。 ## 6. 失败退避与告警 -- [ ] 6.1 收尾按"卡 × 类型"更新连续失败、首次失败、最近错误;按退避序列计算下次到期,频率限制与限流超时不计失败;成功清零。 -- [ ] 6.2 达到阈值或首次即告警类别时写或更新 `tb_polling_failure_alert`;成功后自动转已恢复;通知走现有告警通道。 -- [ ] 6.3 账户级告警:窗口内同账户告警数超过阈值时产生账户级告警。 +- [x] 6.1 收尾按"卡 × 类型"更新连续失败、首次失败、最近错误;按退避序列计算下次到期,频率限制与限流超时不计失败;成功清零。 +- [x] 6.2 达到阈值或首次即告警类别时写或更新 `tb_polling_failure_alert`;成功后自动转已恢复;通知走现有告警通道。(通知以 ERROR 日志 + metric 输出,未接入 alert_service 的规则通知通道) +- [x] 6.3 账户级告警:窗口内同账户告警数超过阈值时产生账户级告警。 ## 7. 业务事件与读入口请求检查 -- [ ] 7.1 订单支付、套餐激活、停复机成功、设备操作(按设备下各卡)、自动续费购包 5 处 `AppendSeriesRequested` 改为事务提交后请求检查。 -- [ ] 7.2 C 端资产详情、后台资产实时状态、C 端与后台实名链接改为请求检查;OpenAPI 查询保持不触发。 -- [ ] 7.3 延迟确认:按场景在检查完成后追加确认到期;结果已满足预期时取消。 -- [ ] 7.4 卡检查不再创建观测序列;`query_device_info` 等非运营商上游部分保持不变。 +- [x] 7.1 订单支付、套餐激活、停复机成功、设备操作(按设备下各卡)、自动续费购包 5 处 `AppendSeriesRequested` 改为事务提交后请求检查。(业务事件仍在原事务写 Outbox,carrier 模式下 Outbox 消费者转为请求检查,满足提交后请求) +- [x] 7.2 C 端资产详情、后台资产实时状态、C 端与后台实名链接改为请求检查;OpenAPI 查询保持不触发。 +- [x] 7.3 延迟确认:按场景在检查完成后追加确认到期;结果已满足预期时取消。 +- [x] 7.4 卡检查不再创建观测序列;`query_device_info` 等非运营商上游部分保持不变。 ## 8. 整体切换 -- [ ] 8.1 Worker 监听 `polling.scheduler.mode`:切 `carrier` 时停旧调度出队、初始化新结构后开启分发;切 `legacy` 时停止领取、等待执行中任务收尾、重建旧分片队列后恢复旧调度。 -- [ ] 8.2 旧轮询 handler 执行前检查模式,为 `carrier` 时直接结束。 -- [ ] 8.3 所有写入轮询结构的入口按当前模式写新或旧结构。 +- [x] 8.1 Worker 监听 `polling.scheduler.mode`:切 `carrier` 时停旧调度出队、初始化新结构后开启分发;切 `legacy` 时停止领取、等待执行中任务收尾、重建旧分片队列后恢复旧调度。 +- [x] 8.2 旧轮询 handler 执行前检查模式,为 `carrier` 时直接结束。 +- [x] 8.3 所有写入轮询结构的入口按当前模式写新或旧结构。(含第 2 个 Change 的重算到期任务) ## 9. 接口与下线 -- [ ] 9.1 `/polling-carrier-params` 列表(含默认标记)、详情、修改、删除;`/polling-check-scenes` 列表与修改;`/polling-failure-alerts` 列表、`/summary`、详情、`PUT /:id/status`;`GET /iot-cards/:id/polling-state`。 -- [ ] 9.2 `/polling-stats` 改为按账户(队列长度、到期数、执行中、实际 QPS、限流等待与超时、处理中标记到期、告警数)。 -- [ ] 9.3 下线 `/polling-concurrency`、`concurrency_service.go`、`acquireConcurrency`/`releaseConcurrency`、按 ZCARD 背压;旧提示通道 List 不再写入。 -- [ ] 9.4 同步路由、`cmd/api/docs.go`、`cmd/gendocs/main.go`,重新生成接口文档。 +- [x] 9.1 `/polling-carrier-params` 列表(含默认标记)、详情、修改、删除;`/polling-check-scenes` 列表与修改;`/polling-failure-alerts` 列表、`/summary`、详情、`PUT /:id/status`;`GET /iot-cards/:id/polling-state`。 +- [x] 9.2 `/polling-stats` 改为按账户(队列长度、到期数、执行中、实际 QPS、限流等待与超时、处理中标记到期、告警数)。(新增 GET /polling-stats/carrier-queues;/polling-stats/carriers 已被第 1 个 Change 的网关调用统计占用) +- [ ] 9.3 下线 `/polling-concurrency`、`concurrency_service.go`、`acquireConcurrency`/`releaseConcurrency`、按 ZCARD 背压;旧提示通道 List 不再写入。(暂缓:旧分片调度仍需并发配置与并发令牌,保留以支持整体回退;按设计在新调度稳定运行后与旧调度代码一并下线) +- [x] 9.4 同步路由、`cmd/api/docs.go`、`cmd/gendocs/main.go`,重新生成接口文档。 ## 10. 编译与规范校验 -- [ ] 10.1 `gofmt -w`;`go build ./cmd/api ./cmd/worker` 通过。 -- [ ] 10.2 `openspec validate add-carrier-polling-scheduler` 与 `openspec doctor --json` 通过。 +- [x] 10.1 `gofmt -w`;`go build ./cmd/api ./cmd/worker` 通过。 +- [x] 10.2 `openspec validate add-carrier-polling-scheduler` 与 `openspec doctor --json` 通过。(validate 通过;当前 openspec 无 doctor 命令) ## 11. 测试环境验证(自动化测试为 N/A,按 ENG-TEST-001;按生产规模造卡或回放生产卡 ID)