From dae44bed30b38bb87ba411f05f95f4b8cd664214 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=82=B9=E8=BF=9B?= Date: Sun, 25 Oct 2026 18:28:42 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8C=89=E8=BF=90=E8=90=A5=E5=95=86?= =?UTF-8?q?=E8=B4=A6=E6=88=B7=E8=B0=83=E5=BA=A6=E7=9A=84=E8=BF=81=E7=A7=BB?= =?UTF-8?q?=E3=80=81=E9=85=8D=E7=BD=AE=E4=B8=8E=E8=B4=A6=E6=88=B7=E9=98=9F?= =?UTF-8?q?=E5=88=97=20Lua?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 迁移 000251~000254：运营商账户调度参数、请求检查场景（预置 8 个）、卡轮询失败告警表，以及调度模式、步骤、退避、告警等受控系统配置（默认 legacy，行为不变）；并发配置表注释标记废弃 - 受控系统配置注册与 JSON 结构校验：步骤只能是 4 种检查类型且不重复、退避序列递增 - yaml 增加 NSQ 地址、msg_timeout、租约、续期间隔、分发间隔、默认执行上限与 max_in_flight；引入 go-nsq - 测试环境 docker-compose 增加 nsqlookupd 与 nsqd，Worker 注入 NSQ 地址 - 账户队列 Redis 结构与 Lua：领取（执行上限、优先占比、按限流组剩余 QPS 估算）、抢执行权、续期、收尾、请求检查（合并、冷却、执行中打优先标记）、移出、发布失败释放 --- docker-compose.prod.yml | 36 ++ go.mod | 2 + go.sum | 4 + internal/bootstrap/handlers.go | 2 + internal/model/carrier_polling.go | 76 ++++ internal/polling/carrierqueue/scripts.go | 247 ++++++++++++ internal/polling/carrierqueue/settings.go | 378 ++++++++++++++++++ internal/polling/carrierqueue/store.go | 278 +++++++++++++ .../000251_add_carrier_polling_param.down.sql | 3 + .../000251_add_carrier_polling_param.up.sql | 42 ++ .../000252_add_polling_check_scene.down.sql | 3 + .../000252_add_polling_check_scene.up.sql | 41 ++ .../000253_add_polling_failure_alert.down.sql | 6 + .../000253_add_polling_failure_alert.up.sql | 46 +++ ...d_carrier_scheduler_system_config.down.sql | 7 + ...add_carrier_scheduler_system_config.up.sql | 31 ++ pkg/config/config.go | 31 ++ pkg/config/defaults/config.yaml | 13 + pkg/config/loader.go | 3 + pkg/constants/carrier_polling.go | 205 ++++++++++ 20 files changed, 1454 insertions(+) create mode 100644 internal/model/carrier_polling.go create mode 100644 internal/polling/carrierqueue/scripts.go create mode 100644 internal/polling/carrierqueue/settings.go create mode 100644 internal/polling/carrierqueue/store.go create mode 100644 migrations/000251_add_carrier_polling_param.down.sql create mode 100644 migrations/000251_add_carrier_polling_param.up.sql create mode 100644 migrations/000252_add_polling_check_scene.down.sql create mode 100644 migrations/000252_add_polling_check_scene.up.sql create mode 100644 migrations/000253_add_polling_failure_alert.down.sql create mode 100644 migrations/000253_add_polling_failure_alert.up.sql create mode 100644 migrations/000254_add_carrier_scheduler_system_config.down.sql create mode 100644 migrations/000254_add_carrier_scheduler_system_config.up.sql create mode 100644 pkg/constants/carrier_polling.go diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 3350435c..e67f9162 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -154,6 +154,9 @@ services: # Worker 执行企微审批提交、同步和恢复任务 - JUNHONG_WECOM_BASE_URL=https://qyapi.weixin.qq.com - JUNHONG_WECOM_TIMEOUT=10s + # 按运营商账户调度的 NSQ（卡轮询分发；调度模式在受控系统配置 polling.scheduler.mode 切换） + - JUNHONG_POLLING_CARRIER_NSQD_ADDRESS=${JUNHONG_POLLING_CARRIER_NSQD_ADDRESS:-nsqd:4150} + - JUNHONG_POLLING_CARRIER_NSQLOOKUPD_ADDRESSES=${JUNHONG_POLLING_CARRIER_NSQLOOKUPD_ADDRESSES:-nsqlookupd:4161} volumes: - ./logs:/app/logs networks: @@ -161,6 +164,39 @@ services: depends_on: api: condition: service_healthy + nsqd: + condition: service_started + logging: + driver: "json-file" + options: + max-size: "10m" + max-file: "3" + + # NSQ：只承担卡轮询的执行分发，Redis 是唯一权威状态（add-carrier-polling-scheduler） + nsqlookupd: + image: nsqio/nsq:v1.3.0 + container_name: junhong-cmp-nsqlookupd + restart: unless-stopped + command: /nsqlookupd + networks: + - junhong-network + logging: + driver: "json-file" + options: + max-size: "10m" + max-file: "3" + + nsqd: + image: nsqio/nsq:v1.3.0 + container_name: junhong-cmp-nsqd + restart: unless-stopped + command: /nsqd --lookupd-tcp-address=nsqlookupd:4160 --broadcast-address=nsqd --data-path=/data --mem-queue-size=10000 --max-msg-timeout=15m + volumes: + - ./nsq-data:/data + networks: + - junhong-network + depends_on: + - nsqlookupd logging: driver: "json-file" options: diff --git a/go.mod b/go.mod index f58dc270..c20890b3 100644 --- a/go.mod +++ b/go.mod @@ -48,6 +48,7 @@ require ( github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-sql-driver/mysql v1.8.1 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect + github.com/golang/snappy v0.0.1 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect @@ -61,6 +62,7 @@ require ( github.com/mattn/go-isatty v0.0.20 // indirect github.com/mattn/go-runewidth v0.0.16 // indirect github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 // indirect + github.com/nsqio/go-nsq v1.1.0 // indirect github.com/patrickmn/go-cache v2.1.0+incompatible // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect diff --git a/go.sum b/go.sum index 79d9f978..d5ac9e55 100644 --- a/go.sum +++ b/go.sum @@ -106,6 +106,8 @@ github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 h1:au07oEsX2xN0kt github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9/go.mod h1:8vg3r2VgvsThLBIFL93Qb5yWzgyZWhEmBwUJWevAkK0= github.com/golang-sql/sqlexp v0.1.0 h1:ZCD6MBpcuOVfGVqsEmY5/4FtYiKz6tSyUv9LPEDei6A= github.com/golang-sql/sqlexp v0.1.0/go.mod h1:J4ad9Vo8ZCWQ2GMrC4UCQy1JpCbwU9m3EOqtpKwwwHI= +github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4= +github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= @@ -175,6 +177,8 @@ github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 h1:RWengNIwukTxcDr9 github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826/go.mod h1:TaXosZuwdSHYgviHp1DAtfrULt5eUgsSMsZf+YrPgl8= github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= +github.com/nsqio/go-nsq v1.1.0 h1:PQg+xxiUjA7V+TLdXw7nVrJ5Jbl3sN86EhGCQj4+FYE= +github.com/nsqio/go-nsq v1.1.0/go.mod h1:vKq36oyeVXgsS5Q8YEO7WghqidAVXQlcFxzQbQTuDEY= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= diff --git a/internal/bootstrap/handlers.go b/internal/bootstrap/handlers.go index 876caf30..cc2811d9 100644 --- a/internal/bootstrap/handlers.go +++ b/internal/bootstrap/handlers.go @@ -29,6 +29,7 @@ import ( systemConfigInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/systemconfig" wecomInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/wecom" pollingPkg "github.com/break/junhong_cmp_fiber/internal/polling" + "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" agentRechargeQuery "github.com/break/junhong_cmp_fiber/internal/query/agentrecharge" assetQuery "github.com/break/junhong_cmp_fiber/internal/query/asset" auditQuery "github.com/break/junhong_cmp_fiber/internal/query/audit" @@ -123,6 +124,7 @@ func initHandlers(svc *services, deps *Dependencies) *Handlers { registerCarrierCallbackConfigDefinitions(systemConfigRegistry, deps.Logger) registerPaymentMethodConfigDefinitions(systemConfigRegistry, deps.Logger) gatewaylimit.RegisterConfigDefinitions(systemConfigRegistry, deps.Logger) + carrierqueue.RegisterConfigDefinitions(systemConfigRegistry, deps.Logger) systemConfigAlerts := systemConfigInfra.NewLogAlertSink(deps.Logger) var systemConfigCache systemConfigInfra.Cache if deps.Redis != nil { diff --git a/internal/model/carrier_polling.go b/internal/model/carrier_polling.go new file mode 100644 index 00000000..2a73e08c --- /dev/null +++ b/internal/model/carrier_polling.go @@ -0,0 +1,76 @@ +package model + +import ( + "time" + + "gorm.io/datatypes" + "gorm.io/gorm" +) + +// CarrierPollingParam 运营商账户调度参数；未配置的账户没有记录，使用全局默认。 +type CarrierPollingParam struct { + ID uint `gorm:"column:id;primaryKey" json:"id"` + CarrierID uint `gorm:"column:carrier_id;not null" json:"carrier_id"` + InflightLimit *int `gorm:"column:inflight_limit" json:"inflight_limit"` + CardMaxInFlight *int `gorm:"column:card_max_in_flight" json:"card_max_in_flight"` + StopResumeMaxInFlight *int `gorm:"column:stop_resume_max_in_flight" json:"stop_resume_max_in_flight"` + SRInflightLimit *int `gorm:"column:sr_inflight_limit" json:"sr_inflight_limit"` + PriorityMaxRatio *float64 `gorm:"column:priority_max_ratio" json:"priority_max_ratio"` + StopResumeMinInterval *int `gorm:"column:stop_resume_min_interval" json:"stop_resume_min_interval"` + StopResumeConfirmDelay *int `gorm:"column:stop_resume_confirm_delay" json:"stop_resume_confirm_delay"` + Creator uint `gorm:"column:creator;not null;default:0" json:"creator"` + Updater uint `gorm:"column:updater;not null;default:0" json:"updater"` + CreatedAt time.Time `gorm:"column:created_at" json:"created_at"` + UpdatedAt time.Time `gorm:"column:updated_at" json:"updated_at"` + DeletedAt gorm.DeletedAt `gorm:"column:deleted_at;index" json:"-"` +} + +// TableName 返回运营商账户调度参数表名。 +func (CarrierPollingParam) TableName() string { + return "tb_carrier_polling_param" +} + +// PollingCheckScene 业务事件与读入口请求检查场景；场景由代码预置，后台只可修改检查类型、确认延迟与冷却。 +type PollingCheckScene struct { + ID uint `gorm:"column:id;primaryKey" json:"id"` + Scene string `gorm:"column:scene;type:varchar(50);not null" json:"scene"` + SceneName string `gorm:"column:scene_name;type:varchar(100);not null" json:"scene_name"` + CheckTypes datatypes.JSON `gorm:"column:check_types;type:jsonb;not null" json:"check_types"` + ConfirmDelays datatypes.JSON `gorm:"column:confirm_delays;type:jsonb;not null" json:"confirm_delays"` + Cooldown *int `gorm:"column:cooldown" json:"cooldown"` + Updater uint `gorm:"column:updater;not null;default:0" json:"updater"` + CreatedAt time.Time `gorm:"column:created_at" json:"created_at"` + UpdatedAt time.Time `gorm:"column:updated_at" json:"updated_at"` +} + +// TableName 返回请求检查场景表名。 +func (PollingCheckScene) TableName() string { + return "tb_polling_check_scene" +} + +// PollingFailureAlert 卡轮询失败告警；同一卡同一检查类型只保留一条未处理告警。 +type PollingFailureAlert struct { + ID uint `gorm:"column:id;primaryKey" json:"id"` + AlertLevel string `gorm:"column:alert_level;type:varchar(20);not null;default:card" json:"alert_level"` + CardID uint `gorm:"column:card_id;not null;default:0" json:"card_id"` + ICCID string `gorm:"column:iccid;type:varchar(30);not null;default:''" json:"iccid"` + CarrierID uint `gorm:"column:carrier_id;not null;default:0" json:"carrier_id"` + CheckType string `gorm:"column:check_type;type:varchar(30);not null" json:"check_type"` + ErrorCategory string `gorm:"column:error_category;type:varchar(50);not null;default:''" json:"error_category"` + FailCount int `gorm:"column:fail_count;not null;default:0" json:"fail_count"` + FirstFailedAt *time.Time `gorm:"column:first_failed_at" json:"first_failed_at"` + LastFailedAt *time.Time `gorm:"column:last_failed_at" json:"last_failed_at"` + LastError string `gorm:"column:last_error;type:varchar(500);not null;default:''" json:"last_error"` + Status int `gorm:"column:status;type:smallint;not null;default:1" json:"status"` + HandledBy uint `gorm:"column:handled_by;not null;default:0" json:"handled_by"` + HandledAt *time.Time `gorm:"column:handled_at" json:"handled_at"` + HandleRemark string `gorm:"column:handle_remark;type:varchar(500);not null;default:''" json:"handle_remark"` + RecoveredAt *time.Time `gorm:"column:recovered_at" json:"recovered_at"` + CreatedAt time.Time `gorm:"column:created_at" json:"created_at"` + UpdatedAt time.Time `gorm:"column:updated_at" json:"updated_at"` +} + +// TableName 返回卡轮询失败告警表名。 +func (PollingFailureAlert) TableName() string { + return "tb_polling_failure_alert" +} diff --git a/internal/polling/carrierqueue/scripts.go b/internal/polling/carrierqueue/scripts.go new file mode 100644 index 00000000..621faa65 --- /dev/null +++ b/internal/polling/carrierqueue/scripts.go @@ -0,0 +1,247 @@ +package carrierqueue + +import "github.com/redis/go-redis/v9" + +// 卡到期 Hash 字段前缀（见 constants.RedisCardDueKey）。 +const ( + fieldDue = "due:" // 各检查类型下次到期（毫秒） + fieldOK = "ok:" // 最近一次成功（毫秒），用于冷却判断 + fieldFail = "fail:" // 连续失败次数 + fieldFirstFail = "ffail:" // 首次失败（毫秒） + fieldError = "err:" // 最近错误（类别|信息） + fieldConfirm = "cfm:" // 待执行的延迟确认序列（秒，逗号分隔） + fieldExpected = "exp:" // 延迟确认的期望结果（如 offline / online），满足时取消后续确认 + fieldPriority = "prio" // 优先标记：1 表示应在优先队列（执行中收到请求检查时由收尾移入优先队列） + fieldCarrier = "carrier" +) + +// claimLua 账户领取（分发器抢到账户分发锁后执行）。 +// +// KEYS[1] 优先队列；KEYS[2] 普通队列；KEYS[3] 执行中计数。 +// ARGV：now_ms、lease_ms、执行上限、优先占比（0 表示不限）、token 前缀、 +// 限流组本秒剩余（"g=n;g=n"，空串表示不按 QPS 限制）、检查类型所属限流组（"type=g;..."）、是否计入实名（1/0）。 +// 返回 {本轮清理的过期执行数, card_id, token, 是否来自优先(1/0), ...}。 +const claimLua = ` +local now = tonumber(ARGV[1]) +local lease = tonumber(ARGV[2]) +local limit = tonumber(ARGV[3]) +local ratio = tonumber(ARGV[4]) +local prefix = ARGV[5] +local unlimited = ARGV[6] == '' +local countRealname = ARGV[8] == '1' + +local expired = redis.call('ZREMRANGEBYSCORE', KEYS[3], '-inf', now) +local result = {expired} +local capacity = limit - redis.call('ZCARD', KEYS[3]) +if capacity <= 0 then + return result +end + +local budget = {} +for g, n in string.gmatch(ARGV[6], '([^=;]+)=(%-?%d+)') do budget[g] = tonumber(n) end +local typeGroup = {} +for t, g in string.gmatch(ARGV[7], '([^=;]+)=([^;]+)') do typeGroup[t] = g end +local gatewayTypes = {'card_status', 'carddata', 'realname'} +local used = {} +local taken = 0 +local stop = false + +local function tryClaim(cardId, fromPriority, qkey) + local dueKey = 'polling:card:' .. cardId .. ':due' + local need = {} + if not unlimited then + for _, t in ipairs(gatewayTypes) do + if t ~= 'realname' or countRealname then + local d = tonumber(redis.call('HGET', dueKey, 'due:' .. t) or '-1') + if d >= 0 and d <= now then + local g = typeGroup[t] + if g then need[g] = (need[g] or 0) + 1 end + end + end + end + for g, n in pairs(need) do + if budget[g] ~= nil and (used[g] or 0) + n > budget[g] then + stop = true + return false + end + end + end + local claimKey = 'polling:card:' .. cardId .. ':claim' + local token = prefix .. ':' .. cardId + local ok = redis.call('SET', claimKey, token, 'NX', 'PX', lease) + if not ok then + local ttl = redis.call('PTTL', claimKey) + if ttl < 0 then ttl = lease end + redis.call('ZADD', qkey, 'XX', now + ttl, cardId) + return false + end + for g, n in pairs(need) do used[g] = (used[g] or 0) + n end + redis.call('ZADD', qkey, 'XX', now + lease, cardId) + redis.call('ZADD', KEYS[3], now + lease, token) + if fromPriority then + redis.call('HDEL', dueKey, 'prio') + end + taken = taken + 1 + table.insert(result, cardId) + table.insert(result, token) + table.insert(result, fromPriority and '1' or '0') + return true +end + +local function drain(qkey, fromPriority, quota) + if quota <= 0 or stop then return end + local ids = redis.call('ZRANGEBYSCORE', qkey, '-inf', now, 'LIMIT', 0, quota * 2 + 10) + for _, id in ipairs(ids) do + if stop or quota <= 0 or taken >= capacity then return end + if tryClaim(id, fromPriority, qkey) then quota = quota - 1 end + end +end + +local priorityQuota = capacity +if ratio > 0 then + priorityQuota = math.floor(capacity * ratio) + if priorityQuota < 1 then priorityQuota = 1 end +end +drain(KEYS[1], true, priorityQuota) +drain(KEYS[2], false, capacity - taken) +drain(KEYS[1], true, capacity - taken) +return result +` + +// renewLua 续期：token 与执行权一致时延长租约与执行权，并推后 score 与执行中计数到期时间。 +// +// KEYS[1] claim；KEYS[2] exec；KEYS[3] 执行中计数；KEYS[4] 优先队列；KEYS[5] 普通队列。 +// ARGV：token、执行权值、lease_ms、exec_ttl_ms、now_ms、card_id。返回 1 成功，0 token 或执行权已不一致。 +const renewLua = ` +if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end +if redis.call('GET', KEYS[2]) ~= ARGV[2] then return 0 end +local lease = tonumber(ARGV[3]) +local now = tonumber(ARGV[5]) +redis.call('PEXPIRE', KEYS[1], lease) +redis.call('PEXPIRE', KEYS[2], tonumber(ARGV[4])) +redis.call('ZADD', KEYS[3], 'XX', now + lease, ARGV[1]) +redis.call('ZADD', KEYS[4], 'XX', now + lease, ARGV[6]) +redis.call('ZADD', KEYS[5], 'XX', now + lease, ARGV[6]) +return 1 +` + +// acquireExecLua 抢执行权：claim 等于 token 时 SET exec NX PX。返回 1 抢到，0 claim 不一致，-1 已有执行者。 +// +// KEYS[1] claim；KEYS[2] exec。ARGV：token、执行权值、exec_ttl_ms。 +const acquireExecLua = ` +if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end +if redis.call('SET', KEYS[2], ARGV[2], 'NX', 'PX', tonumber(ARGV[3])) then return 1 end +return -1 +` + +// finishLua 收尾：写回业务结果；token 一致时删除处理中标记与执行权、移出执行中计数， +// 按优先标记把卡放入优先或普通队列，score = 各检查类型最早到期。卡已不在任一队列（已移出轮询）时不重新加入。 +// +// KEYS[1] due；KEYS[2] claim；KEYS[3] exec；KEYS[4] 执行中计数；KEYS[5] 优先队列；KEYS[6] 普通队列。 +// ARGV：token、card_id、now_ms、set 字段对数 n、字段1、值1、…、删除字段…。 +// 返回 1 正常收尾，0 token 不一致（只写了业务结果）。 +const finishLua = ` +local n = tonumber(ARGV[4]) +local idx = 5 +for i = 1, n do + redis.call('HSET', KEYS[1], ARGV[idx], ARGV[idx + 1]) + idx = idx + 2 +end +while idx <= #ARGV do + redis.call('HDEL', KEYS[1], ARGV[idx]) + idx = idx + 1 +end +if redis.call('GET', KEYS[2]) ~= ARGV[1] then return 0 end +redis.call('DEL', KEYS[2]) +redis.call('DEL', KEYS[3]) +redis.call('ZREM', KEYS[4], ARGV[1]) +local cardId = ARGV[2] +local inPriority = redis.call('ZSCORE', KEYS[5], cardId) +local inQueue = redis.call('ZSCORE', KEYS[6], cardId) +if not inPriority and not inQueue then return 1 end +local fields = redis.call('HGETALL', KEYS[1]) +local nextDue = nil +for i = 1, #fields, 2 do + if string.sub(fields[i], 1, 4) == 'due:' then + local v = tonumber(fields[i + 1]) + if v and (nextDue == nil or v < nextDue) then nextDue = v end + end +end +if nextDue == nil then nextDue = tonumber(ARGV[3]) + 3600000 end +if redis.call('HGET', KEYS[1], 'prio') == '1' then + redis.call('ZREM', KEYS[6], cardId) + redis.call('ZADD', KEYS[5], nextDue, cardId) +else + redis.call('ZREM', KEYS[5], cardId) + redis.call('ZADD', KEYS[6], nextDue, cardId) +end +return 1 +` + +// requestCheckLua 请求检查：把指定检查类型设为立即到期并放入优先队列；卡执行中时只打优先标记，收尾后立即再执行。 +// 只作用于仍在轮询中的卡（在优先或普通队列中）。冷却期内（最近成功距今未超过冷却）的类型不请求。 +// +// KEYS[1] due；KEYS[2] 优先队列；KEYS[3] 普通队列；KEYS[4] claim。 +// ARGV：now_ms、card_id、cooldown_ms（0 不冷却）、确认延迟（秒，逗号分隔，可空）、期望结果（可空）、类型…。 +// 返回本次实际请求的类型数；卡不在轮询中返回 -1。 +const requestCheckLua = ` +local now = tonumber(ARGV[1]) +local cardId = ARGV[2] +local cooldown = tonumber(ARGV[3]) +local inPriority = redis.call('ZSCORE', KEYS[2], cardId) +local inQueue = redis.call('ZSCORE', KEYS[3], cardId) +if not inPriority and not inQueue then return -1 end +local requested = 0 +for i = 6, #ARGV do + local t = ARGV[i] + local skip = false + if cooldown > 0 then + local ok = tonumber(redis.call('HGET', KEYS[1], 'ok:' .. t) or '0') + if ok > 0 and now - ok < cooldown then skip = true end + end + if not skip then + redis.call('HSET', KEYS[1], 'due:' .. t, now) + if ARGV[4] ~= '' then + redis.call('HSET', KEYS[1], 'cfm:' .. t, ARGV[4]) + if ARGV[5] ~= '' then redis.call('HSET', KEYS[1], 'exp:' .. t, ARGV[5]) else redis.call('HDEL', KEYS[1], 'exp:' .. t) end + end + requested = requested + 1 + end +end +if requested == 0 then return 0 end +redis.call('HSET', KEYS[1], 'prio', '1') +if redis.call('EXISTS', KEYS[4]) == 1 then return requested end +redis.call('ZREM', KEYS[3], cardId) +redis.call('ZADD', KEYS[2], now, cardId) +return requested +` + +// removeLua 把卡移出所属账户队列（关闭轮询、删除卡等）；执行中的卡收尾时因不在队列中而不会被重新加入。 +// KEYS[1] 优先队列；KEYS[2] 普通队列；KEYS[3] due。ARGV：card_id。 +const removeLua = ` +redis.call('ZREM', KEYS[1], ARGV[1]) +redis.call('ZREM', KEYS[2], ARGV[1]) +redis.call('DEL', KEYS[3]) +return 1 +` + +// releaseLua 发布失败时按 token 删除处理中标记、移出执行中计数，并把 score 恢复为当前时间。 +// KEYS[1] claim；KEYS[2] 执行中计数；KEYS[3] 优先队列；KEYS[4] 普通队列。ARGV：token、card_id、now_ms。 +const releaseLua = ` +if redis.call('GET', KEYS[1]) ~= ARGV[1] then return 0 end +redis.call('DEL', KEYS[1]) +redis.call('ZREM', KEYS[2], ARGV[1]) +redis.call('ZADD', KEYS[3], 'XX', tonumber(ARGV[3]), ARGV[2]) +redis.call('ZADD', KEYS[4], 'XX', tonumber(ARGV[3]), ARGV[2]) +return 1 +` + +var ( + claimScript = redis.NewScript(claimLua) + renewScript = redis.NewScript(renewLua) + acquireExecScript = redis.NewScript(acquireExecLua) + finishScript = redis.NewScript(finishLua) + requestCheckScript = redis.NewScript(requestCheckLua) + removeScript = redis.NewScript(removeLua) + releaseScript = redis.NewScript(releaseLua) +) diff --git a/internal/polling/carrierqueue/settings.go b/internal/polling/carrierqueue/settings.go new file mode 100644 index 00000000..7fc01232 --- /dev/null +++ b/internal/polling/carrierqueue/settings.go @@ -0,0 +1,378 @@ +// Package carrierqueue 实现按运营商账户调度的卡轮询结构：账户延时队列、卡到期信息、领取 / 续期 / 收尾 / 请求检查 Lua、 +// 分发器、初始化与受控系统配置。Redis 是唯一权威状态，NSQ 只做传输（add-carrier-polling-scheduler）。 +package carrierqueue + +import ( + "context" + "encoding/json" + stderrors "errors" + "strconv" + "sync" + "time" + + "go.uber.org/zap" + + "github.com/break/junhong_cmp_fiber/internal/infrastructure/systemconfig" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// settingsTTL 是调度参数在进程内的缓存时长；后台修改数秒内生效。 +const settingsTTL = 5 * time.Second + +// RegisterConfigDefinitions 注册按运营商账户调度的受控系统配置定义（含 JSON 结构校验）。 +func RegisterConfigDefinitions(registry *systemconfig.Registry, logger *zap.Logger) { + if registry == nil { + return + } + minOne, maxSeconds, maxCount := int64(1), int64(86400*7), int64(100000) + definitions := []systemconfig.Definition{ + {Key: constants.SystemConfigPollingSchedulerMode, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeString, + DefaultValue: constants.PollingSchedulerModeLegacy, Description: "卡轮询调度模式：legacy 旧分片调度，carrier 按运营商账户调度（整体切换）", + EnumValues: []string{constants.PollingSchedulerModeLegacy, constants.PollingSchedulerModeCarrier}}, + {Key: constants.SystemConfigPollingStepOrder, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, + DefaultValue: `["card_status","carddata","realname","package"]`, Description: "卡任务步骤顺序", Validator: validateStepOrder}, + {Key: constants.SystemConfigPollingStepTimeout, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, + DefaultValue: `{"card_status":15,"carddata":15,"realname":15,"package":20}`, Description: "各检查类型单步超时（秒）", Validator: validateStepTimeout}, + {Key: constants.SystemConfigPollingTaskTimeoutMargin, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt, + DefaultValue: "10", Description: "卡任务总超时余量（秒）", Min: &minOne, Max: &maxSeconds}, + {Key: constants.SystemConfigPollingSkipPackageAfterFlow, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeBool, + DefaultValue: "true", Description: "同轮流量步骤已触发停复机评估时跳过套餐步骤"}, + {Key: constants.SystemConfigPollingFailureBackoff, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, + DefaultValue: `{"default":[60,300,900,1800,3600]}`, Description: "失败退避序列（秒，递增），可按检查类型覆盖 default", Validator: validateBackoff}, + {Key: constants.SystemConfigPollingAlertFailCount, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, + DefaultValue: `{"default":5}`, Description: "连续失败告警阈值，可按检查类型覆盖 default", Validator: validateAlertCount}, + {Key: constants.SystemConfigPollingAlertFailDuration, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt, + DefaultValue: "21600", Description: "持续失败时长告警阈值（秒）", Min: &minOne, Max: &maxSeconds}, + {Key: constants.SystemConfigPollingImmediateAlert, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeJSON, + DefaultValue: `["account_config"]`, Description: "首次出现即告警的错误类别", Validator: validateCategories}, + {Key: constants.SystemConfigPollingAccountAlertWindow, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt, + DefaultValue: "600", Description: "账户级告警统计窗口（秒）", Min: &minOne, Max: &maxSeconds}, + {Key: constants.SystemConfigPollingAccountAlertThreshold, Module: constants.SystemConfigModulePolling, ValueType: constants.SystemConfigTypeInt, + DefaultValue: "50", Description: "账户级告警阈值：窗口内同账户新增失败告警数", Min: &minOne, Max: &maxCount}, + } + for _, definition := range definitions { + if _, exists := registry.Get(definition.Key); exists { + continue + } + if err := registry.Register(definition); err != nil && logger != nil { + logger.Error("注册按运营商账户调度系统配置失败，将使用代码默认值", zap.String("config_key", definition.Key), zap.Error(err)) + } + } +} + +func validCheckType(value string) bool { + switch value { + case constants.CheckTypeCardStatus, constants.CheckTypeCarddata, constants.CheckTypeRealname, constants.CheckTypePackage: + return true + } + return false +} + +func validateStepOrder(raw string) error { + var steps []string + if err := json.Unmarshal([]byte(raw), &steps); err != nil { + return stderrors.New("步骤顺序必须是字符串数组") + } + if len(steps) == 0 { + return stderrors.New("步骤顺序不能为空") + } + seen := map[string]bool{} + for _, step := range steps { + if !validCheckType(step) { + return stderrors.New("步骤只能是 card_status、carddata、realname、package") + } + if seen[step] { + return stderrors.New("步骤不能重复") + } + seen[step] = true + } + return nil +} + +func validateStepTimeout(raw string) error { + var timeouts map[string]int + if err := json.Unmarshal([]byte(raw), &timeouts); err != nil { + return stderrors.New("单步超时必须是 {检查类型: 秒} 对象") + } + for key, value := range timeouts { + if !validCheckType(key) { + return stderrors.New("单步超时的检查类型非法") + } + if value <= 0 || value > 600 { + return stderrors.New("单步超时必须在 1~600 秒之间") + } + } + return nil +} + +func validateBackoff(raw string) error { + var backoff map[string][]int + if err := json.Unmarshal([]byte(raw), &backoff); err != nil { + return stderrors.New("退避序列必须是 {default 或检查类型: [秒...]} 对象") + } + if len(backoff["default"]) == 0 { + return stderrors.New("退避序列必须包含非空的 default") + } + for key, values := range backoff { + if key != "default" && !validCheckType(key) { + return stderrors.New("退避序列的检查类型非法") + } + for index, value := range values { + if value <= 0 { + return stderrors.New("退避间隔必须为正数") + } + if index > 0 && value <= values[index-1] { + return stderrors.New("退避序列必须递增") + } + } + } + return nil +} + +func validateAlertCount(raw string) error { + var counts map[string]int + if err := json.Unmarshal([]byte(raw), &counts); err != nil { + return stderrors.New("告警阈值必须是 {default 或检查类型: 次数} 对象") + } + if counts["default"] <= 0 { + return stderrors.New("告警阈值必须包含正数 default") + } + for key, value := range counts { + if key != "default" && !validCheckType(key) { + return stderrors.New("告警阈值的检查类型非法") + } + if value <= 0 { + return stderrors.New("告警阈值必须为正数") + } + } + return nil +} + +func validateCategories(raw string) error { + var categories []string + if err := json.Unmarshal([]byte(raw), &categories); err != nil { + return stderrors.New("错误类别必须是字符串数组") + } + return nil +} + +// Settings 是按运营商账户调度的运行参数快照。 +type Settings struct { + Mode string + StepOrder []string + StepTimeouts map[string]time.Duration + TaskTimeoutMargin time.Duration + SkipPackageAfterFlow bool + Backoff map[string][]time.Duration + AlertFailCount map[string]int + AlertFailDuration time.Duration + ImmediateAlert map[string]bool + AccountAlertWindow time.Duration + AccountAlertLimit int +} + +// CarrierMode 判断当前是否为按运营商账户调度。 +func (s Settings) CarrierMode() bool { + return s.Mode == constants.PollingSchedulerModeCarrier +} + +// StepTimeout 返回检查类型单步超时；未配置时 15 秒。 +func (s Settings) StepTimeout(checkType string) time.Duration { + if value, ok := s.StepTimeouts[checkType]; ok && value > 0 { + return value + } + return 15 * time.Second +} + +// TaskTimeout 返回卡任务总超时 = 各步骤超时之和 + 余量。 +func (s Settings) TaskTimeout() time.Duration { + total := s.TaskTimeoutMargin + for _, step := range s.StepOrder { + total += s.StepTimeout(step) + } + return total +} + +// BackoffFor 返回检查类型第 n 次连续失败（n 从 1 起）的退避间隔，不超过 maxInterval（> 0 时）。 +func (s Settings) BackoffFor(checkType string, failCount int, maxInterval time.Duration) time.Duration { + sequence := s.Backoff[checkType] + if len(sequence) == 0 { + sequence = s.Backoff["default"] + } + if len(sequence) == 0 { + sequence = []time.Duration{time.Minute} + } + index := failCount - 1 + if index < 0 { + index = 0 + } + if index >= len(sequence) { + index = len(sequence) - 1 + } + delay := sequence[index] + if maxInterval > 0 && delay > maxInterval { + delay = maxInterval + } + return delay +} + +// AlertThreshold 返回检查类型连续失败告警阈值。 +func (s Settings) AlertThreshold(checkType string) int { + if value, ok := s.AlertFailCount[checkType]; ok && value > 0 { + return value + } + if value := s.AlertFailCount["default"]; value > 0 { + return value + } + return 5 +} + +// DefaultSettings 是配置读取失败时的安全默认值（旧分片调度）。 +func DefaultSettings() Settings { + return Settings{ + Mode: constants.PollingSchedulerModeLegacy, + StepOrder: constants.AllCheckTypes(), + StepTimeouts: map[string]time.Duration{ + constants.CheckTypeCardStatus: 15 * time.Second, constants.CheckTypeCarddata: 15 * time.Second, + constants.CheckTypeRealname: 15 * time.Second, constants.CheckTypePackage: 20 * time.Second, + }, + TaskTimeoutMargin: 10 * time.Second, SkipPackageAfterFlow: true, + Backoff: map[string][]time.Duration{"default": {time.Minute, 5 * time.Minute, 15 * time.Minute, 30 * time.Minute, time.Hour}}, + AlertFailCount: map[string]int{"default": 5}, AlertFailDuration: 6 * time.Hour, + ImmediateAlert: map[string]bool{constants.GatewayClassAccountConfig: true}, + AccountAlertWindow: 10 * time.Minute, AccountAlertLimit: 50, + } +} + +// ConfigReader 读取已注册的受控系统配置。 +type ConfigReader interface { + Get(ctx context.Context, key string) (string, error) +} + +// SettingsProvider 在进程内缓存调度参数，超过 TTL 重新读取。 +type SettingsProvider struct { + reader ConfigReader + logger *zap.Logger + mu sync.Mutex + cached Settings + loaded time.Time +} + +// NewSettingsProvider 创建调度参数提供者；reader 为 nil 时始终返回默认值。 +func NewSettingsProvider(reader ConfigReader, logger *zap.Logger) *SettingsProvider { + if logger == nil { + logger = zap.NewNop() + } + return &SettingsProvider{reader: reader, logger: logger, cached: DefaultSettings()} +} + +// Current 返回当前调度参数；读取失败时保留上一次成功值。 +func (p *SettingsProvider) Current(ctx context.Context) Settings { + if p == nil { + return DefaultSettings() + } + p.mu.Lock() + defer p.mu.Unlock() + if p.reader == nil || time.Since(p.loaded) < settingsTTL { + return p.cached + } + next, err := p.load(ctx) + p.loaded = time.Now() + if err != nil { + p.logger.Warn("读取按运营商账户调度配置失败，沿用上一次配置", zap.Error(err)) + return p.cached + } + p.cached = next + return next +} + +// Invalidate 使缓存立即失效（后台保存后调用）。 +func (p *SettingsProvider) Invalidate() { + if p == nil { + return + } + p.mu.Lock() + p.loaded = time.Time{} + p.mu.Unlock() +} + +func (p *SettingsProvider) load(ctx context.Context) (Settings, error) { + s := DefaultSettings() + get := func(key string) (string, error) { return p.reader.Get(ctx, key) } + raw, err := get(constants.SystemConfigPollingSchedulerMode) + if err != nil { + return s, err + } + s.Mode = raw + if raw, err = get(constants.SystemConfigPollingStepOrder); err != nil { + return s, err + } + if err := json.Unmarshal([]byte(raw), &s.StepOrder); err != nil || validateStepOrder(raw) != nil { + return s, stderrors.New("步骤顺序配置非法") + } + if raw, err = get(constants.SystemConfigPollingStepTimeout); err != nil { + return s, err + } + var timeouts map[string]int + if err := json.Unmarshal([]byte(raw), &timeouts); err != nil { + return s, stderrors.New("单步超时配置非法") + } + for key, value := range timeouts { + s.StepTimeouts[key] = time.Duration(value) * time.Second + } + if raw, err = get(constants.SystemConfigPollingFailureBackoff); err != nil { + return s, err + } + var backoff map[string][]int + if err := json.Unmarshal([]byte(raw), &backoff); err != nil { + return s, stderrors.New("退避序列配置非法") + } + s.Backoff = map[string][]time.Duration{} + for key, values := range backoff { + for _, value := range values { + s.Backoff[key] = append(s.Backoff[key], time.Duration(value)*time.Second) + } + } + if raw, err = get(constants.SystemConfigPollingAlertFailCount); err != nil { + return s, err + } + if err := json.Unmarshal([]byte(raw), &s.AlertFailCount); err != nil { + return s, stderrors.New("告警阈值配置非法") + } + if raw, err = get(constants.SystemConfigPollingImmediateAlert); err != nil { + return s, err + } + var categories []string + if err := json.Unmarshal([]byte(raw), &categories); err != nil { + return s, stderrors.New("首次告警类别配置非法") + } + s.ImmediateAlert = map[string]bool{} + for _, category := range categories { + s.ImmediateAlert[category] = true + } + if raw, err = get(constants.SystemConfigPollingSkipPackageAfterFlow); err != nil { + return s, err + } + s.SkipPackageAfterFlow = raw == "true" + ints := []struct { + key string + set func(int) + }{ + {constants.SystemConfigPollingTaskTimeoutMargin, func(v int) { s.TaskTimeoutMargin = time.Duration(v) * time.Second }}, + {constants.SystemConfigPollingAlertFailDuration, func(v int) { s.AlertFailDuration = time.Duration(v) * time.Second }}, + {constants.SystemConfigPollingAccountAlertWindow, func(v int) { s.AccountAlertWindow = time.Duration(v) * time.Second }}, + {constants.SystemConfigPollingAccountAlertThreshold, func(v int) { s.AccountAlertLimit = v }}, + } + for _, item := range ints { + raw, err := get(item.key) + if err != nil { + return s, err + } + value, convErr := strconv.Atoi(raw) + if convErr != nil { + return s, stderrors.New("调度整数配置非法") + } + item.set(value) + } + return s, nil +} diff --git a/internal/polling/carrierqueue/store.go b/internal/polling/carrierqueue/store.go new file mode 100644 index 00000000..f4c086e0 --- /dev/null +++ b/internal/polling/carrierqueue/store.go @@ -0,0 +1,278 @@ +package carrierqueue + +import ( + "context" + stderrors "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// Store 封装按运营商账户调度的 Redis 结构与 Lua 原子操作。 +type Store struct { + redis *redis.Client +} + +// NewStore 创建账户队列存储。 +func NewStore(redisClient *redis.Client) *Store { + return &Store{redis: redisClient} +} + +// Client 返回底层 Redis 客户端。 +func (s *Store) Client() *redis.Client { + return s.redis +} + +// ClaimInput 是一次账户领取的参数。 +type ClaimInput struct { + CarrierID uint + Lease time.Duration + InflightLimit int + PriorityRatio float64 // 0 表示不限 + TokenPrefix string // 本轮派发 token 前缀（实例 + 轮次），token = 前缀:card_id + GroupBudget map[string]int // 限流组本秒剩余；nil 表示不按 QPS 限制 + TypeGroup map[string]string // 检查类型 → 限流组 + CountRealname bool // 预计调用数是否计入实名（账户未开实名管控时不计） +} + +// Claimed 是一张被领取的卡。 +type Claimed struct { + CardID uint + Token string + FromPriority bool +} + +// Claim 执行一次账户领取，返回领取到的卡与本轮清理的过期执行数（处理中崩溃或超时次数）。 +func (s *Store) Claim(ctx context.Context, input ClaimInput) ([]Claimed, int64, error) { + budget := "" + if input.GroupBudget != nil { + parts := make([]string, 0, len(input.GroupBudget)) + for group, remaining := range input.GroupBudget { + parts = append(parts, group+"="+strconv.Itoa(remaining)) + } + budget = strings.Join(parts, ";") + if budget == "" { + budget = "none=0" + } + } + typeParts := make([]string, 0, len(input.TypeGroup)) + for checkType, group := range input.TypeGroup { + typeParts = append(typeParts, checkType+"="+group) + } + countRealname := "0" + if input.CountRealname { + countRealname = "1" + } + raw, err := claimScript.Run(ctx, s.redis, + []string{constants.RedisCarrierPriorityQueueKey(input.CarrierID), constants.RedisCarrierQueueKey(input.CarrierID), constants.RedisCarrierRunningKey(input.CarrierID)}, + time.Now().UnixMilli(), input.Lease.Milliseconds(), input.InflightLimit, strconv.FormatFloat(input.PriorityRatio, 'f', 3, 64), + input.TokenPrefix, budget, strings.Join(typeParts, ";"), countRealname, + ).Slice() + if err != nil { + return nil, 0, err + } + if len(raw) == 0 { + return nil, 0, nil + } + expired, _ := raw[0].(int64) + claimed := make([]Claimed, 0, (len(raw)-1)/3) + for index := 1; index+2 < len(raw); index += 3 { + cardID, _ := strconv.ParseUint(fmt.Sprint(raw[index]), 10, 64) + claimed = append(claimed, Claimed{CardID: uint(cardID), Token: fmt.Sprint(raw[index+1]), FromPriority: fmt.Sprint(raw[index+2]) == "1"}) + } + return claimed, expired, nil +} + +// AcquireExec 抢执行权：返回 1 抢到，0 处理中标记已不是该 token（过期或重复投递），-1 该 token 已有执行者。 +func (s *Store) AcquireExec(ctx context.Context, cardID uint, token, execValue string, ttl time.Duration) (int64, error) { + return acquireExecScript.Run(ctx, s.redis, []string{constants.RedisCardClaimKey(cardID), constants.RedisCardExecKey(cardID)}, + token, execValue, ttl.Milliseconds()).Int64() +} + +// Renew 续期租约与执行权；返回 false 表示 token 或执行权已不一致，调用方必须停止后续步骤。 +func (s *Store) Renew(ctx context.Context, cardID, carrierID uint, token, execValue string, lease, execTTL time.Duration) (bool, error) { + result, err := renewScript.Run(ctx, s.redis, + []string{constants.RedisCardClaimKey(cardID), constants.RedisCardExecKey(cardID), constants.RedisCarrierRunningKey(carrierID), + constants.RedisCarrierPriorityQueueKey(carrierID), constants.RedisCarrierQueueKey(carrierID)}, + token, execValue, lease.Milliseconds(), execTTL.Milliseconds(), time.Now().UnixMilli(), cardID).Int64() + return result == 1, err +} + +// Finish 收尾：写回业务结果（set 字段与删除字段），token 一致时释放处理中标记并按最早到期设置 score。 +// 返回 false 表示 token 已不一致，只写入了业务结果。 +func (s *Store) Finish(ctx context.Context, cardID, carrierID uint, token string, set map[string]string, del []string) (bool, error) { + args := []any{token, cardID, time.Now().UnixMilli(), len(set)} + for field, value := range set { + args = append(args, field, value) + } + for _, field := range del { + args = append(args, field) + } + result, err := finishScript.Run(ctx, s.redis, + []string{constants.RedisCardDueKey(cardID), constants.RedisCardClaimKey(cardID), constants.RedisCardExecKey(cardID), + constants.RedisCarrierRunningKey(carrierID), constants.RedisCarrierPriorityQueueKey(carrierID), constants.RedisCarrierQueueKey(carrierID)}, + args...).Int64() + return result == 1, err +} + +// Release 发布失败时按 token 删除处理中标记并把 score 恢复为当前时间。 +func (s *Store) Release(ctx context.Context, cardID, carrierID uint, token string) error { + return releaseScript.Run(ctx, s.redis, + []string{constants.RedisCardClaimKey(cardID), constants.RedisCarrierRunningKey(carrierID), + constants.RedisCarrierPriorityQueueKey(carrierID), constants.RedisCarrierQueueKey(carrierID)}, + token, cardID, time.Now().UnixMilli()).Err() +} + +// CheckRequest 是一次请求检查。 +type CheckRequest struct { + CardID uint + CarrierID uint + CheckTypes []string + Cooldown time.Duration + ConfirmDelays []time.Duration + Expected string // 延迟确认的期望结果（card_status 为 online / offline），满足时取消后续确认 +} + +// RequestCheck 把卡的指定检查类型设为立即到期放入优先队列；返回实际请求的类型数，卡不在轮询中返回 -1。 +func (s *Store) RequestCheck(ctx context.Context, request CheckRequest) (int64, error) { + if len(request.CheckTypes) == 0 { + return 0, nil + } + confirm := make([]string, 0, len(request.ConfirmDelays)) + for _, delay := range request.ConfirmDelays { + confirm = append(confirm, strconv.FormatInt(int64(delay/time.Second), 10)) + } + args := []any{time.Now().UnixMilli(), request.CardID, request.Cooldown.Milliseconds(), strings.Join(confirm, ","), request.Expected} + for _, checkType := range request.CheckTypes { + args = append(args, checkType) + } + return requestCheckScript.Run(ctx, s.redis, + []string{constants.RedisCardDueKey(request.CardID), constants.RedisCarrierPriorityQueueKey(request.CarrierID), + constants.RedisCarrierQueueKey(request.CarrierID), constants.RedisCardClaimKey(request.CardID)}, + args...).Int64() +} + +// Remove 把卡移出所属账户队列并删除到期信息（关闭轮询、删除卡）。 +func (s *Store) Remove(ctx context.Context, cardID, carrierID uint) error { + return removeScript.Run(ctx, s.redis, + []string{constants.RedisCarrierPriorityQueueKey(carrierID), constants.RedisCarrierQueueKey(carrierID), constants.RedisCardDueKey(cardID)}, + cardID).Err() +} + +// Upsert 写入卡各检查类型到期时间并按最早到期放入账户队列（已在优先队列中的保留在优先队列）。 +// 卡所属账户变化时先移出旧账户队列。dues 为空表示该卡没有任何需要执行的检查类型，移出队列。 +func (s *Store) Upsert(ctx context.Context, cardID, carrierID uint, dues map[string]time.Time) error { + dueKey := constants.RedisCardDueKey(cardID) + previous, err := s.redis.HGet(ctx, dueKey, fieldCarrier).Uint64() + if err != nil && !stderrors.Is(err, redis.Nil) { + return err + } + pipe := s.redis.TxPipeline() + if previous != 0 && uint(previous) != carrierID { + pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(uint(previous)), cardID) + pipe.ZRem(ctx, constants.RedisCarrierQueueKey(uint(previous)), cardID) + } + if len(dues) == 0 { + pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(carrierID), cardID) + pipe.ZRem(ctx, constants.RedisCarrierQueueKey(carrierID), cardID) + pipe.Del(ctx, dueKey) + _, err := pipe.Exec(ctx) + return err + } + values := map[string]any{fieldCarrier: carrierID} + var earliest time.Time + for _, checkType := range constants.AllCheckTypes() { + due, ok := dues[checkType] + if !ok { + pipe.HDel(ctx, dueKey, fieldDue+checkType) + continue + } + values[fieldDue+checkType] = due.UnixMilli() + if earliest.IsZero() || due.Before(earliest) { + earliest = due + } + } + pipe.HSet(ctx, dueKey, values) + _, err = pipe.Exec(ctx) + if err != nil { + return err + } + priorityKey := constants.RedisCarrierPriorityQueueKey(carrierID) + if inPriority, _ := s.redis.ZScore(ctx, priorityKey, strconv.FormatUint(uint64(cardID), 10)).Result(); inPriority > 0 { + return s.redis.ZAdd(ctx, priorityKey, redis.Z{Score: float64(earliest.UnixMilli()), Member: cardID}).Err() + } + return s.redis.ZAdd(ctx, constants.RedisCarrierQueueKey(carrierID), redis.Z{Score: float64(earliest.UnixMilli()), Member: cardID}).Err() +} + +// CardDue 是卡到期信息的解析结果。 +type CardDue struct { + CarrierID uint + Priority bool + Types map[string]TypeDue +} + +// TypeDue 是卡某检查类型的到期与失败信息。 +type TypeDue struct { + Due time.Time + HasDue bool + LastOK time.Time + FailCount int + FirstFailed time.Time + LastError string + Confirm []time.Duration + Expected string +} + +// ReadDue 读取并解析卡到期信息；Hash 不存在时返回空结构。 +func (s *Store) ReadDue(ctx context.Context, cardID uint) (CardDue, error) { + raw, err := s.redis.HGetAll(ctx, constants.RedisCardDueKey(cardID)).Result() + if err != nil { + return CardDue{}, err + } + return ParseDue(raw), nil +} + +// ParseDue 解析卡到期 Hash。 +func ParseDue(raw map[string]string) CardDue { + result := CardDue{Types: map[string]TypeDue{}} + if value, err := strconv.ParseUint(raw[fieldCarrier], 10, 64); err == nil { + result.CarrierID = uint(value) + } + result.Priority = raw[fieldPriority] == "1" + for _, checkType := range constants.AllCheckTypes() { + item := TypeDue{} + if ms, err := strconv.ParseInt(raw[fieldDue+checkType], 10, 64); err == nil { + item.Due, item.HasDue = time.UnixMilli(ms), true + } + if ms, err := strconv.ParseInt(raw[fieldOK+checkType], 10, 64); err == nil && ms > 0 { + item.LastOK = time.UnixMilli(ms) + } + item.FailCount, _ = strconv.Atoi(raw[fieldFail+checkType]) + if ms, err := strconv.ParseInt(raw[fieldFirstFail+checkType], 10, 64); err == nil && ms > 0 { + item.FirstFailed = time.UnixMilli(ms) + } + item.LastError = raw[fieldError+checkType] + for _, part := range strings.Split(raw[fieldConfirm+checkType], ",") { + if seconds, err := strconv.Atoi(strings.TrimSpace(part)); err == nil && seconds > 0 { + item.Confirm = append(item.Confirm, time.Duration(seconds)*time.Second) + } + } + item.Expected = raw[fieldExpected+checkType] + result.Types[checkType] = item + } + return result +} + +// Field 名称导出给执行器组装收尾写入。 +func FieldDue(checkType string) string { return fieldDue + checkType } +func FieldOK(checkType string) string { return fieldOK + checkType } +func FieldFail(checkType string) string { return fieldFail + checkType } +func FieldFirstFail(checkType string) string { return fieldFirstFail + checkType } +func FieldError(checkType string) string { return fieldError + checkType } +func FieldConfirm(checkType string) string { return fieldConfirm + checkType } +func FieldExpected(checkType string) string { return fieldExpected + checkType } diff --git a/migrations/000251_add_carrier_polling_param.down.sql b/migrations/000251_add_carrier_polling_param.down.sql new file mode 100644 index 00000000..a5045ed3 --- /dev/null +++ b/migrations/000251_add_carrier_polling_param.down.sql @@ -0,0 +1,3 @@ +-- 回滚运营商账户调度参数表。 +DROP INDEX IF EXISTS uk_carrier_polling_param_carrier; +DROP TABLE IF EXISTS tb_carrier_polling_param; diff --git a/migrations/000251_add_carrier_polling_param.up.sql b/migrations/000251_add_carrier_polling_param.up.sql new file mode 100644 index 00000000..86e824a5 --- /dev/null +++ b/migrations/000251_add_carrier_polling_param.up.sql @@ -0,0 +1,42 @@ +-- 运营商账户调度参数（add-carrier-polling-scheduler）。 +-- 背景：新调度按运营商账户领取与执行卡任务，每个账户的执行上限、每进程并发、优先占比需要可单独配置。 +-- +-- 设计选择： +-- 1. carrier_id 唯一，一个账户一行；未配置的账户没有记录、使用 yaml 全局默认，不改 tb_carrier。 +-- 2. 各参数可空，空表示该项使用全局默认。 +-- 3. stop_resume_* 与 sr_inflight_limit 供停复机指令队列（add-stop-resume-command-queue）使用，本次只建列。 +-- 4. 不使用数据库外键（ENG-MODEL-001）；软删除，后台维护。 + +CREATE TABLE tb_carrier_polling_param ( + id BIGSERIAL PRIMARY KEY, + carrier_id BIGINT NOT NULL, + inflight_limit INTEGER, + card_max_in_flight INTEGER, + stop_resume_max_in_flight INTEGER, + sr_inflight_limit INTEGER, + priority_max_ratio NUMERIC(4, 3), + stop_resume_min_interval INTEGER, + stop_resume_confirm_delay INTEGER, + creator BIGINT NOT NULL DEFAULT 0, + updater BIGINT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + deleted_at TIMESTAMPTZ, + CONSTRAINT ck_carrier_polling_param_positive CHECK ( + (inflight_limit IS NULL OR inflight_limit > 0) AND (card_max_in_flight IS NULL OR card_max_in_flight > 0) + AND (stop_resume_max_in_flight IS NULL OR stop_resume_max_in_flight > 0) AND (sr_inflight_limit IS NULL OR sr_inflight_limit > 0) + AND (stop_resume_min_interval IS NULL OR stop_resume_min_interval > 0) AND (stop_resume_confirm_delay IS NULL OR stop_resume_confirm_delay > 0)), + CONSTRAINT ck_carrier_polling_param_ratio CHECK (priority_max_ratio IS NULL OR (priority_max_ratio > 0 AND priority_max_ratio <= 1)) +); + +CREATE UNIQUE INDEX uk_carrier_polling_param_carrier ON tb_carrier_polling_param (carrier_id) WHERE deleted_at IS NULL; + +COMMENT ON TABLE tb_carrier_polling_param IS '运营商账户调度参数：未配置的账户使用全局默认'; +COMMENT ON COLUMN tb_carrier_polling_param.carrier_id IS '运营商账户ID'; +COMMENT ON COLUMN tb_carrier_polling_param.inflight_limit IS '该账户同时执行卡任务的上限（跨 Worker 实例）'; +COMMENT ON COLUMN tb_carrier_polling_param.card_max_in_flight IS '每个 Worker 进程消费该账户卡任务 topic 的 max_in_flight'; +COMMENT ON COLUMN tb_carrier_polling_param.stop_resume_max_in_flight IS '每个 Worker 进程消费该账户停复机 topic 的 max_in_flight'; +COMMENT ON COLUMN tb_carrier_polling_param.sr_inflight_limit IS '该账户同时执行停复机指令的上限'; +COMMENT ON COLUMN tb_carrier_polling_param.priority_max_ratio IS '卡任务每轮领取中优先队列的最大占比（0~1），空表示不限'; +COMMENT ON COLUMN tb_carrier_polling_param.stop_resume_min_interval IS '单卡两次停复机网关调用的最小间隔（秒）'; +COMMENT ON COLUMN tb_carrier_polling_param.stop_resume_confirm_delay IS '停复机结果未知后查卡状态确认的延迟（秒）'; diff --git a/migrations/000252_add_polling_check_scene.down.sql b/migrations/000252_add_polling_check_scene.down.sql new file mode 100644 index 00000000..04c4bb89 --- /dev/null +++ b/migrations/000252_add_polling_check_scene.down.sql @@ -0,0 +1,3 @@ +-- 回滚业务事件与读入口检查场景表。 +DROP INDEX IF EXISTS uk_polling_check_scene_scene; +DROP TABLE IF EXISTS tb_polling_check_scene; diff --git a/migrations/000252_add_polling_check_scene.up.sql b/migrations/000252_add_polling_check_scene.up.sql new file mode 100644 index 00000000..e2b00ed7 --- /dev/null +++ b/migrations/000252_add_polling_check_scene.up.sql @@ -0,0 +1,41 @@ +-- 业务事件与读入口检查场景（add-carrier-polling-scheduler）。 +-- 背景：业务事件与读入口原先各自起"立即、3 分钟、5 分钟"卡观测序列，不受限流、与优先轮询重复； +-- 改为"请求检查"：把卡的对应检查类型设为立即到期放入运营商账户优先队列，按场景配置确认延迟与冷却。 +-- +-- 设计选择： +-- 1. scene 唯一，场景由代码预置；后台只可修改检查类型、确认延迟与冷却时长。 +-- 2. check_types 取值 card_status / carddata / realname；confirm_delays 为秒数数组（空数组表示只立即检查一次）； +-- cooldown 为秒数，空表示不冷却（读入口场景默认配置冷却，避免反复刷新占满优先队列）。 +-- 3. 停复机成功与设备操作默认 3 分钟后确认一次（结果已满足预期时取消），沿用原观测序列的确认语义。 + +CREATE TABLE tb_polling_check_scene ( + id BIGSERIAL PRIMARY KEY, + scene VARCHAR(50) NOT NULL, + scene_name VARCHAR(100) NOT NULL, + check_types JSONB NOT NULL DEFAULT '[]'::jsonb, + confirm_delays JSONB NOT NULL DEFAULT '[]'::jsonb, + cooldown INTEGER, + updater BIGINT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT ck_polling_check_scene_cooldown CHECK (cooldown IS NULL OR cooldown > 0) +); + +CREATE UNIQUE INDEX uk_polling_check_scene_scene ON tb_polling_check_scene (scene); + +COMMENT ON TABLE tb_polling_check_scene IS '业务事件与读入口请求检查场景'; +COMMENT ON COLUMN tb_polling_check_scene.scene IS '场景编码（代码预置）'; +COMMENT ON COLUMN tb_polling_check_scene.scene_name IS '场景名称'; +COMMENT ON COLUMN tb_polling_check_scene.check_types IS '需要检查的类型：card_status / carddata / realname'; +COMMENT ON COLUMN tb_polling_check_scene.confirm_delays IS '延迟确认序列（秒），空数组表示只立即检查一次'; +COMMENT ON COLUMN tb_polling_check_scene.cooldown IS '冷却时长（秒）：最近一次成功检查未超过该时长则不请求，空表示不冷却'; + +INSERT INTO tb_polling_check_scene (scene, scene_name, check_types, confirm_delays, cooldown) VALUES + ('order_paid', '订单支付', '["card_status","carddata","realname"]', '[]', NULL), + ('package_activated', '套餐激活', '["card_status","carddata","realname"]', '[]', NULL), + ('stop_resume_succeeded', '停复机成功', '["card_status"]', '[180]', NULL), + ('device_operation', '设备操作', '["card_status"]', '[180]', NULL), + ('auto_renewal_purchase', '自动续费购包', '["card_status","carddata","realname"]', '[]', NULL), + ('client_asset_detail', 'C 端资产详情', '["card_status","carddata"]', '[]', 300), + ('admin_asset_realtime', '后台资产实时状态', '["card_status","carddata","realname"]', '[]', 60), + ('realname_link', '实名链接', '["realname"]', '[]', 60); diff --git a/migrations/000253_add_polling_failure_alert.down.sql b/migrations/000253_add_polling_failure_alert.down.sql new file mode 100644 index 00000000..ec32d56c --- /dev/null +++ b/migrations/000253_add_polling_failure_alert.down.sql @@ -0,0 +1,6 @@ +-- 回滚卡轮询失败告警表。 +DROP INDEX IF EXISTS idx_polling_failure_alert_created; +DROP INDEX IF EXISTS idx_polling_failure_alert_card; +DROP INDEX IF EXISTS idx_polling_failure_alert_carrier; +DROP INDEX IF EXISTS uk_polling_failure_alert_open; +DROP TABLE IF EXISTS tb_polling_failure_alert; diff --git a/migrations/000253_add_polling_failure_alert.up.sql b/migrations/000253_add_polling_failure_alert.up.sql new file mode 100644 index 00000000..7e6dccc5 --- /dev/null +++ b/migrations/000253_add_polling_failure_alert.up.sql @@ -0,0 +1,46 @@ +-- 卡轮询失败告警（add-carrier-polling-scheduler）。 +-- 背景：新调度按"卡 × 检查类型"记连续失败并退避重试；持续失败需要告警，由运营联系上游或关闭该卡轮询。 +-- +-- 设计选择： +-- 1. 同一卡同一检查类型只保留一条未处理告警（部分唯一索引），后续失败更新计数与最近错误。 +-- 2. status：1-未处理 2-已恢复（首次成功自动转入）3-已关闭轮询 4-已处理（人工标记）。 +-- 3. check_type 取值 card_status / carddata / realname / package；停复机方向在停复机指令队列接入同一张表。 +-- 4. alert_level：card 卡级 / carrier 账户级（同账户窗口内告警数超阈值时产生，card_id 为 0）。 +-- 5. 告警写入走异步，失败只记日志，不影响轮询；不使用数据库外键（ENG-MODEL-001）。 + +CREATE TABLE tb_polling_failure_alert ( + id BIGSERIAL PRIMARY KEY, + alert_level VARCHAR(20) NOT NULL DEFAULT 'card', + card_id BIGINT NOT NULL DEFAULT 0, + iccid VARCHAR(30) NOT NULL DEFAULT '', + carrier_id BIGINT NOT NULL DEFAULT 0, + check_type VARCHAR(30) NOT NULL, + error_category VARCHAR(50) NOT NULL DEFAULT '', + fail_count INTEGER NOT NULL DEFAULT 0, + first_failed_at TIMESTAMPTZ, + last_failed_at TIMESTAMPTZ, + last_error VARCHAR(500) NOT NULL DEFAULT '', + status SMALLINT NOT NULL DEFAULT 1, + handled_by BIGINT NOT NULL DEFAULT 0, + handled_at TIMESTAMPTZ, + handle_remark VARCHAR(500) NOT NULL DEFAULT '', + recovered_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT ck_polling_failure_alert_status CHECK (status IN (1, 2, 3, 4)), + CONSTRAINT ck_polling_failure_alert_level CHECK (alert_level IN ('card', 'carrier')) +); + +CREATE UNIQUE INDEX uk_polling_failure_alert_open ON tb_polling_failure_alert (alert_level, card_id, carrier_id, check_type) WHERE status = 1; +CREATE INDEX idx_polling_failure_alert_carrier ON tb_polling_failure_alert (carrier_id, status, last_failed_at DESC); +CREATE INDEX idx_polling_failure_alert_card ON tb_polling_failure_alert (card_id, created_at DESC); +CREATE INDEX idx_polling_failure_alert_created ON tb_polling_failure_alert (created_at DESC); + +COMMENT ON TABLE tb_polling_failure_alert IS '卡轮询失败告警：同一卡同一检查类型只保留一条未处理告警'; +COMMENT ON COLUMN tb_polling_failure_alert.alert_level IS '告警级别 card-卡级 carrier-账户级'; +COMMENT ON COLUMN tb_polling_failure_alert.card_id IS '卡ID，账户级告警为 0'; +COMMENT ON COLUMN tb_polling_failure_alert.check_type IS '检查类型 card_status / carddata / realname / package；账户级告警为 all'; +COMMENT ON COLUMN tb_polling_failure_alert.error_category IS '最近错误类别（网关错误类别或上游错误分类）'; +COMMENT ON COLUMN tb_polling_failure_alert.fail_count IS '连续失败次数；账户级告警为窗口内告警数'; +COMMENT ON COLUMN tb_polling_failure_alert.last_error IS '最近错误（脱敏截断）'; +COMMENT ON COLUMN tb_polling_failure_alert.status IS '状态 1-未处理 2-已恢复 3-已关闭轮询 4-已处理'; diff --git a/migrations/000254_add_carrier_scheduler_system_config.down.sql b/migrations/000254_add_carrier_scheduler_system_config.down.sql new file mode 100644 index 00000000..ecb39023 --- /dev/null +++ b/migrations/000254_add_carrier_scheduler_system_config.down.sql @@ -0,0 +1,7 @@ +-- 回滚按运营商账户调度的受控系统配置与并发配置表注释。 +DELETE FROM tb_system_config WHERE config_key IN ( + 'polling.scheduler.mode', 'polling.scheduler.step_order', 'polling.scheduler.step_timeout_seconds', + 'polling.scheduler.task_timeout_margin_seconds', 'polling.scheduler.skip_package_after_flow', + 'polling.failure.backoff_seconds', 'polling.failure.alert_fail_count', 'polling.failure.alert_fail_duration_seconds', + 'polling.failure.immediate_alert_categories', 'polling.failure.account_alert_window_seconds', 'polling.failure.account_alert_threshold'); +COMMENT ON TABLE tb_polling_concurrency_config IS '并发控制配置表 - 控制不同类型任务的最大并发数'; diff --git a/migrations/000254_add_carrier_scheduler_system_config.up.sql b/migrations/000254_add_carrier_scheduler_system_config.up.sql new file mode 100644 index 00000000..3c2994f4 --- /dev/null +++ b/migrations/000254_add_carrier_scheduler_system_config.up.sql @@ -0,0 +1,31 @@ +-- 按运营商账户调度的受控系统配置（add-carrier-polling-scheduler），module = polling。 +-- 配置的权威定义（类型、取值范围、默认值、JSON 结构校验）在代码注册表中；本迁移只落库初始值。 +-- 已存在同名 Key 时不覆盖（ON CONFLICT DO NOTHING），避免覆盖运营已修改的值。 +-- +-- 1. polling.scheduler.mode：legacy 旧分片调度（默认）/ carrier 按运营商账户调度；全局整体切换，不按账户灰度。 +-- 2. polling.scheduler.step_order：卡任务步骤顺序，只能是 card_status / carddata / realname / package 且不重复。 +-- 3. polling.scheduler.step_timeout_seconds：各检查类型单步超时（秒）。 +-- 4. polling.scheduler.task_timeout_margin_seconds：卡任务总超时 = 各步骤超时之和 + 余量。 +-- 5. polling.scheduler.skip_package_after_flow：同轮流量步骤已触发停复机评估时跳过套餐步骤。 +-- 6. polling.failure.backoff_seconds：失败退避序列（秒，递增），可按检查类型覆盖 default；最长不超过该类型正常间隔。 +-- 7. polling.failure.alert_fail_count：连续失败告警阈值，可按检查类型覆盖 default。 +-- 8. polling.failure.alert_fail_duration_seconds：持续失败时长告警阈值（秒）。 +-- 9. polling.failure.immediate_alert_categories：首次出现即告警的错误类别。 +-- 10. polling.failure.account_alert_window_seconds / account_alert_threshold：账户级告警窗口与阈值。 +-- 同时把 tb_polling_concurrency_config 表注释标记为废弃（新调度按账户执行上限与限流取代全局并发配置），不删表。 + +INSERT INTO tb_system_config (config_key, config_value, value_type, module, description, is_readonly, is_sensitive, creator, updater, created_at, updated_at) VALUES + ('polling.scheduler.mode', 'legacy', 'string', 'polling', '卡轮询调度模式：legacy 旧分片调度，carrier 按运营商账户调度（整体切换）', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.scheduler.step_order', '["card_status","carddata","realname","package"]', 'json', 'polling', '卡任务步骤顺序', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.scheduler.step_timeout_seconds', '{"card_status":15,"carddata":15,"realname":15,"package":20}', 'json', 'polling', '各检查类型单步超时（秒）', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.scheduler.task_timeout_margin_seconds', '10', 'int', 'polling', '卡任务总超时余量（秒）：总超时 = 各步骤超时之和 + 余量', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.scheduler.skip_package_after_flow', 'true', 'bool', 'polling', '同轮流量步骤已触发停复机评估时跳过套餐步骤', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.failure.backoff_seconds', '{"default":[60,300,900,1800,3600]}', 'json', 'polling', '失败退避序列（秒，递增），可按检查类型覆盖 default', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.failure.alert_fail_count', '{"default":5}', 'json', 'polling', '连续失败告警阈值，可按检查类型覆盖 default', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.failure.alert_fail_duration_seconds', '21600', 'int', 'polling', '持续失败时长告警阈值（秒）', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.failure.immediate_alert_categories', '["account_config"]', 'json', 'polling', '首次出现即告警的错误类别', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.failure.account_alert_window_seconds', '600', 'int', 'polling', '账户级告警统计窗口（秒）', FALSE, FALSE, 0, 0, NOW(), NOW()), + ('polling.failure.account_alert_threshold', '50', 'int', 'polling', '账户级告警阈值：窗口内同账户新增失败告警数', FALSE, FALSE, 0, 0, NOW(), NOW()) +ON CONFLICT (config_key) DO NOTHING; + +COMMENT ON TABLE tb_polling_concurrency_config IS '已废弃：按运营商账户调度后由账户执行上限与网关限流取代，仅旧分片调度使用'; diff --git a/pkg/config/config.go b/pkg/config/config.go index 54c8a953..50f663dc 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -213,6 +213,37 @@ type PollingConfig struct { // VerboseLog 开启后，所有轮询 handler 在成功获取上游数据时以 Info 级别输出详细日志 // 环境变量：JUNHONG_POLLING_VERBOSE_LOG VerboseLog bool `mapstructure:"verbose_log"` + // Carrier 按运营商账户调度（add-carrier-polling-scheduler）的部署参数；调度模式本身放受控系统配置 polling.scheduler.mode。 + Carrier CarrierSchedulerConfig `mapstructure:"carrier"` +} + +// CarrierSchedulerConfig 按运营商账户调度的 yaml 参数。 +// 约束：msg_timeout > renew_interval；lease > msg_timeout × max_attempts；执行权 TTL = renew_interval × 2。 +type CarrierSchedulerConfig struct { + // NSQDAddress 分发器发布消息的 nsqd TCP 地址，如 nsqd:4150；为空时新调度不可用（切到 carrier 会告警并保持不领取）。 + // 环境变量：JUNHONG_POLLING_CARRIER_NSQD_ADDRESS + NSQDAddress string `mapstructure:"nsqd_address"` + // NSQLookupdAddresses 消费者发现 nsqd 的 nsqlookupd HTTP 地址，逗号分隔，如 nsqlookupd:4161；为空时消费者直连 NSQDAddress。 + // 环境变量：JUNHONG_POLLING_CARRIER_NSQLOOKUPD_ADDRESSES + NSQLookupdAddresses string `mapstructure:"nsqlookupd_addresses"` + // MsgTimeout NSQ 消息超时，须大于续期间隔，执行中靠 Touch 延长。 + MsgTimeout time.Duration `mapstructure:"msg_timeout"` + // MaxAttempts NSQ 允许的投递次数（含首次）。 + MaxAttempts uint16 `mapstructure:"max_attempts"` + // Lease 处理中标记 TTL（租约），须大于 msg_timeout × max_attempts。 + Lease time.Duration `mapstructure:"lease"` + // RenewInterval 租约续期间隔，默认租约 / 3 以内。 + RenewInterval time.Duration `mapstructure:"renew_interval"` + // DispatchInterval 分发间隔。 + DispatchInterval time.Duration `mapstructure:"dispatch_interval"` + // ShutdownGrace 优雅退出宽限期：停止分发与订阅后等待执行中任务完成的时长。 + ShutdownGrace time.Duration `mapstructure:"shutdown_grace"` + // DefaultInflightLimit 未配置调度参数的账户同时执行卡任务的上限（跨实例）。 + DefaultInflightLimit int `mapstructure:"default_inflight_limit"` + // DefaultCardMaxInFlight 未配置调度参数的账户每进程消费卡任务的 max_in_flight。 + DefaultCardMaxInFlight int `mapstructure:"default_card_max_in_flight"` + // SubscriberProcesses 订阅卡任务 topic 的 Worker 进程数，用于校验执行上限不超过各进程 max_in_flight 之和。 + SubscriberProcesses int `mapstructure:"subscriber_processes"` } // PollingAutoTriggerConfig 轮询自动触发配置 diff --git a/pkg/config/defaults/config.yaml b/pkg/config/defaults/config.yaml index fd068aed..7691535b 100644 --- a/pkg/config/defaults/config.yaml +++ b/pkg/config/defaults/config.yaml @@ -133,6 +133,19 @@ gateway: # 轮询配置 polling: verbose_log: false # 是否开启详细日志（环境变量：JUNHONG_POLLING_VERBOSE_LOG） + # 按运营商账户调度（调度模式在受控系统配置 polling.scheduler.mode 切换） + carrier: + nsqd_address: "" # nsqd TCP 地址，如 nsqd:4150（环境变量：JUNHONG_POLLING_CARRIER_NSQD_ADDRESS） + nsqlookupd_addresses: "" # nsqlookupd HTTP 地址，逗号分隔（环境变量：JUNHONG_POLLING_CARRIER_NSQLOOKUPD_ADDRESSES） + msg_timeout: 90s # NSQ 消息超时，须大于续期间隔 + max_attempts: 3 # NSQ 允许投递次数（含首次） + lease: 300s # 处理中标记 TTL，须大于 msg_timeout × max_attempts + renew_interval: 30s # 租约续期间隔 + dispatch_interval: 1s # 分发间隔 + shutdown_grace: 30s # 优雅退出宽限期 + default_inflight_limit: 20 # 未配置账户的执行上限（跨实例） + default_card_max_in_flight: 20 # 未配置账户的每进程 max_in_flight + subscriber_processes: 1 # 订阅卡任务的 Worker 进程数（校验执行上限用） # 轮询自动触发配置 polling_auto_trigger: diff --git a/pkg/config/loader.go b/pkg/config/loader.go index 884845e5..3bab6cb7 100644 --- a/pkg/config/loader.go +++ b/pkg/config/loader.go @@ -131,6 +131,9 @@ func bindEnvVariables(v *viper.Viper) { "gateway.app_secret", "gateway.timeout", "polling.verbose_log", + "polling.carrier.nsqd_address", + "polling.carrier.nsqlookupd_addresses", + "polling.carrier.subscriber_processes", "polling_auto_trigger.enable_auto_trigger", "polling_auto_trigger.auto_trigger_system_user_id", "worker.role", diff --git a/pkg/constants/carrier_polling.go b/pkg/constants/carrier_polling.go new file mode 100644 index 00000000..bd38c1cb --- /dev/null +++ b/pkg/constants/carrier_polling.go @@ -0,0 +1,205 @@ +package constants + +import "fmt" + +// 卡轮询调度模式（受控系统配置 polling.scheduler.mode），全局整体切换，不按账户灰度。 +const ( + PollingSchedulerModeLegacy = "legacy" // 旧分片调度（Asynq） + PollingSchedulerModeCarrier = "carrier" // 按运营商账户调度（Redis 延时队列 + NSQ） +) + +// 卡任务检查类型（新调度内部与配置使用的短名）。 +const ( + CheckTypeCardStatus = "card_status" + CheckTypeCarddata = "carddata" + CheckTypeRealname = "realname" + CheckTypePackage = "package" +) + +// AllCheckTypes 返回全部检查类型（默认步骤顺序）。返回新切片。 +func AllCheckTypes() []string { + return []string{CheckTypeCardStatus, CheckTypeCarddata, CheckTypeRealname, CheckTypePackage} +} + +// CheckTypeFromTaskType 把旧任务类型（polling:xxx）转为检查类型短名；不认识返回空串。 +func CheckTypeFromTaskType(taskType string) string { + switch taskType { + case TaskTypePollingCardStatus: + return CheckTypeCardStatus + case TaskTypePollingCarddata: + return CheckTypeCarddata + case TaskTypePollingRealname: + return CheckTypeRealname + case TaskTypePollingPackage: + return CheckTypePackage + default: + return "" + } +} + +// TaskTypeFromCheckType 把检查类型短名转为旧任务类型（polling:xxx）；不认识返回空串。 +func TaskTypeFromCheckType(checkType string) string { + switch checkType { + case CheckTypeCardStatus: + return TaskTypePollingCardStatus + case CheckTypeCarddata: + return TaskTypePollingCarddata + case CheckTypeRealname: + return TaskTypePollingRealname + case CheckTypePackage: + return TaskTypePollingPackage + default: + return "" + } +} + +// CheckTypeName 返回检查类型中文名称。 +func CheckTypeName(checkType string) string { + switch checkType { + case CheckTypeCardStatus: + return "卡状态检查" + case CheckTypeCarddata: + return "流量检查" + case CheckTypeRealname: + return "实名检查" + case CheckTypePackage: + return "套餐检查" + case PollingAlertCheckTypeAll: + return "全部" + default: + return checkType + } +} + +// CheckTypeCallsGateway 判断检查类型是否调用运营商网关（套餐检查只按规则判断停复机，不调网关）。 +func CheckTypeCallsGateway(checkType string) bool { + return checkType == CheckTypeCardStatus || checkType == CheckTypeCarddata || checkType == CheckTypeRealname +} + +// 请求检查场景（tb_polling_check_scene.scene，代码预置）。 +const ( + PollingCheckSceneOrderPaid = "order_paid" + PollingCheckScenePackageActivated = "package_activated" + PollingCheckSceneStopResumeSucceeded = "stop_resume_succeeded" + PollingCheckSceneDeviceOperation = "device_operation" + PollingCheckSceneAutoRenewalPurchase = "auto_renewal_purchase" + PollingCheckSceneClientAssetDetail = "client_asset_detail" + PollingCheckSceneAdminAssetRealtime = "admin_asset_realtime" + PollingCheckSceneRealnameLink = "realname_link" +) + +// 失败告警级别与状态（tb_polling_failure_alert）。 +const ( + PollingAlertLevelCard = "card" + PollingAlertLevelCarrier = "carrier" + // PollingAlertCheckTypeAll 是账户级告警的检查类型。 + PollingAlertCheckTypeAll = "all" + + PollingAlertStatusOpen = 1 // 未处理 + PollingAlertStatusRecovered = 2 // 已恢复（首次成功自动转入） + PollingAlertStatusPollingDisabled = 3 // 已关闭轮询 + PollingAlertStatusHandled = 4 // 已处理（人工标记） +) + +// PollingAlertStatusName 返回失败告警状态名称。 +func PollingAlertStatusName(status int) string { + switch status { + case PollingAlertStatusOpen: + return "未处理" + case PollingAlertStatusRecovered: + return "已恢复" + case PollingAlertStatusPollingDisabled: + return "已关闭轮询" + case PollingAlertStatusHandled: + return "已处理" + default: + return "" + } +} + +// 失败类别中不计入连续失败的取值（限流等待超时与上游频率限制）。 +const ( + // PollingFailureRateLimited 本账户限流等待超时（请求未发出）。 + PollingFailureRateLimited = "rate_limited" +) + +// 按运营商账户调度的受控系统配置键（module = polling）。 +const ( + SystemConfigPollingSchedulerMode = "polling.scheduler.mode" + SystemConfigPollingStepOrder = "polling.scheduler.step_order" + SystemConfigPollingStepTimeout = "polling.scheduler.step_timeout_seconds" + SystemConfigPollingTaskTimeoutMargin = "polling.scheduler.task_timeout_margin_seconds" + SystemConfigPollingSkipPackageAfterFlow = "polling.scheduler.skip_package_after_flow" + SystemConfigPollingFailureBackoff = "polling.failure.backoff_seconds" + SystemConfigPollingAlertFailCount = "polling.failure.alert_fail_count" + SystemConfigPollingAlertFailDuration = "polling.failure.alert_fail_duration_seconds" + SystemConfigPollingImmediateAlert = "polling.failure.immediate_alert_categories" + SystemConfigPollingAccountAlertWindow = "polling.failure.account_alert_window_seconds" + SystemConfigPollingAccountAlertThreshold = "polling.failure.account_alert_threshold" +) + +// ======================================== +// 按运营商账户调度的 Redis Key（前缀与旧分片调度不冲突） +// ======================================== + +// RedisCarrierPriorityQueueKey 账户优先队列（ZSET，member=card_id，score=可执行时间毫秒）。 +func RedisCarrierPriorityQueueKey(carrierID uint) string { + return fmt.Sprintf("polling:carrier:%d:priority", carrierID) +} + +// RedisCarrierQueueKey 账户普通队列（ZSET，member=card_id，score=该卡最早到期时间毫秒）。 +func RedisCarrierQueueKey(carrierID uint) string { + return fmt.Sprintf("polling:carrier:%d:queue", carrierID) +} + +// RedisCarrierRunningKey 账户执行中计数（ZSET，member=派发 token，score=租约到期毫秒），只用于执行上限与监控。 +func RedisCarrierRunningKey(carrierID uint) string { + return fmt.Sprintf("polling:carrier:%d:running", carrierID) +} + +// RedisCarrierDispatchLockKey 账户分发锁：同一账户每轮只由一个分发器实例领取。 +func RedisCarrierDispatchLockKey(carrierID uint) string { + return fmt.Sprintf("polling:carrier:%d:dispatch_lock", carrierID) +} + +// RedisCarrierExpiredClaimsKey 账户处理中标记到期次数（Hash，field=分钟，value=次数），供监控。 +func RedisCarrierExpiredClaimsKey(carrierID uint) string { + return fmt.Sprintf("polling:carrier:%d:expired_claims", carrierID) +} + +// RedisCardDueKey 卡到期信息（Hash）：各检查类型 due:{type}（毫秒）、fail:{type}、first_fail:{type}、 +// err:{type}、ok:{type}（最近成功毫秒）、confirm:{type}（待确认延迟序列）、carrier（所属账户）、priority（优先标记）。 +func RedisCardDueKey(cardID uint) string { + return fmt.Sprintf("polling:card:%d:due", cardID) +} + +// RedisCardClaimKey 卡任务处理中标记，值=派发 token，TTL=租约。 +func RedisCardClaimKey(cardID uint) string { + return fmt.Sprintf("polling:card:%d:claim", cardID) +} + +// RedisCardExecKey 卡任务执行权，值=派发 token + 消费者实例，TTL=续期间隔×2。 +func RedisCardExecKey(cardID uint) string { + return fmt.Sprintf("polling:card:%d:exec", cardID) +} + +// RedisCarrierSchedulerHeartbeatKey 分发器心跳（Hash，field=实例，value=毫秒）。 +const RedisCarrierSchedulerHeartbeatKey = "polling:carrier:dispatcher:heartbeat" + +// RedisCarrierSchedulerInitStateKey 新调度初始化状态（String：running / done），初始化完成前分发器不领取。 +const RedisCarrierSchedulerInitStateKey = "polling:carrier:init_state" + +// RedisCarrierAccountAlertKey 账户级告警窗口计数（ZSET，member=告警ID，score=毫秒）。 +func RedisCarrierAccountAlertKey(carrierID uint) string { + return fmt.Sprintf("polling:carrier:%d:alert_window", carrierID) +} + +// NSQ topic 与 channel。 +const ( + NSQChannelWorker = "worker" +) + +// NSQCardTopic 返回账户卡任务 topic。 +func NSQCardTopic(carrierID uint) string { + return fmt.Sprintf("polling.card.%d", carrierID) +}