feat: 按运营商账户调度的迁移、配置与账户队列 Lua

- 迁移 000251~000254：运营商账户调度参数、请求检查场景（预置 8 个）、卡轮询失败告警表，以及调度模式、步骤、退避、告警等受控系统配置（默认 legacy，行为不变）；并发配置表注释标记废弃 - 受控系统配置注册与 JSON 结构校验：步骤只能是 4 种检查类型且不重复、退避序列递增 - yaml 增加 NSQ 地址、msg_timeout、租约、续期间隔、分发间隔、默认执行上限与 max_in_flight；引入 go-nsq - 测试环境 docker-compose 增加 nsqlookupd 与 nsqd，Worker 注入 NSQ 地址 - 账户队列 Redis 结构与 Lua：领取（执行上限、优先占比、按限流组剩余 QPS 估算）、抢执行权、续期、收尾、请求检查（合并、冷却、执行中打优先标记）、移出、发布失败释放
2026-10-25 18:28:42 +09:00
parent f49baff7c2
commit dae44bed30
20 changed files with 1454 additions and 0 deletions
--- 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:
--- 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
--- 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=
--- 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 {
--- a/internal/model/carrier_polling.go
+++ 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"
+}
--- a/internal/polling/carrierqueue/scripts.go
+++ 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)
+)
--- a/internal/polling/carrierqueue/settings.go
+++ 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
+}
--- a/internal/polling/carrierqueue/store.go
+++ 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 }
--- a/migrations/000251_add_carrier_polling_param.down.sql
+++ 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;
--- a/migrations/000251_add_carrier_polling_param.up.sql
+++ 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 '停复机结果未知后查卡状态确认的延迟（秒）';
--- a/migrations/000252_add_polling_check_scene.down.sql
+++ 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;
--- a/migrations/000252_add_polling_check_scene.up.sql
+++ 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);
--- a/migrations/000253_add_polling_failure_alert.down.sql
+++ 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;
--- a/migrations/000253_add_polling_failure_alert.up.sql
+++ 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-已处理';
--- a/migrations/000254_add_carrier_scheduler_system_config.down.sql
+++ 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 '并发控制配置表 - 控制不同类型任务的最大并发数';
--- a/migrations/000254_add_carrier_scheduler_system_config.up.sql
+++ 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 '已废弃：按运营商账户调度后由账户执行上限与网关限流取代，仅旧分片调度使用';
--- 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 轮询自动触发配置
--- 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:
--- 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",
--- a/pkg/constants/carrier_polling.go
+++ 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)
+}