diff --git a/cmd/worker/carrier_scheduler.go b/cmd/worker/carrier_scheduler.go new file mode 100644 index 00000000..07af0487 --- /dev/null +++ b/cmd/worker/carrier_scheduler.go @@ -0,0 +1,228 @@ +package main + +import ( + "context" + "os" + "strconv" + "sync" + "time" + + "go.uber.org/zap" + + "github.com/break/junhong_cmp_fiber/internal/infrastructure/nsqbus" + "github.com/break/junhong_cmp_fiber/internal/polling" + "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" + "github.com/break/junhong_cmp_fiber/internal/task" + "github.com/break/junhong_cmp_fiber/pkg/config" + "github.com/break/junhong_cmp_fiber/pkg/constants" + "github.com/break/junhong_cmp_fiber/pkg/queue" +) + +// carrierInitLockKey 新结构初始化锁：多实例时只由一个实例初始化。 +const carrierInitLockKey = "polling:carrier:init_lock" + +// carrierSchedulerRuntime 是按运营商账户调度在 Worker 内的运行组件：模式监听、初始化、分发器与 NSQ 消费者。 +type carrierSchedulerRuntime struct { + cfg config.CarrierSchedulerConfig + runtime *workerRuntime + initializer *polling.CarrierInitializer + dispatcher *carrierqueue.Dispatcher + consumers *nsqbus.ConsumerManager + producer *nsqbus.Producer + alerts *carrierqueue.AlertWriter + executor *task.CarrierCardExecutor + singleton bool + instanceID string + logger *zap.Logger + + mu sync.Mutex + lastMode string + reinitQueued bool + legacyRestarts func() +} + +// newCarrierSchedulerParts 在 Worker 运行时初始化阶段创建调度参数、账户队列存储与模式接缝， +// 并注入旧调度相关组件（生命周期、轮询基类），使其在启动前即可按模式写入正确结构。 +func newCarrierSchedulerParts(runtime *workerRuntime, appLogger *zap.Logger) { + settings := carrierqueue.NewDefaultSettingsProvider(runtime.db, runtime.redisClient, appLogger) + store := carrierqueue.NewStore(runtime.redisClient) + scheduling := polling.NewCarrierScheduling(store, settings, runtime.pollingConfigMgr, appLogger) + runtime.carrierSettings = settings + runtime.carrierStore = store + runtime.carrierScheduling = scheduling + runtime.lifecycleSvc.SetCarrierScheduling(scheduling) + runtime.pollingBase.SetSchedulerModeReader(func(ctx context.Context) string { return settings.Current(ctx).Mode }) + if runtime.gatewayLimit != nil && runtime.gatewayLimit.Classifier != nil { + classifier := runtime.gatewayLimit.Classifier + runtime.pollingBase.SetErrorClassifier(func(ctx context.Context, op, message string) string { + return classifier.Classify(ctx, op, message).Category + }) + } +} + +// startCarrierScheduler 启动模式监听；分发器只在 leader/all 运行（多实例安全），消费者在所有处理任务的实例运行。 +func startCarrierScheduler(ctx context.Context, cfg *config.Config, runtime *workerRuntime, steps queue.PollingStepHandlers, + singleton bool, legacyRestart func(), appLogger *zap.Logger) *carrierSchedulerRuntime { + carrierCfg := cfg.Polling.Carrier + instanceID := cfg.Worker.InstanceName + if instanceID == "" { + instanceID, _ = os.Hostname() + } + instanceID += "-" + strconv.Itoa(os.Getpid()) + alerts := carrierqueue.NewAlertWriter(runtime.db, runtime.redisClient, runtime.carrierSettings, appLogger) + alerts.Start(ctx) + executor := task.NewCarrierCardExecutor(runtime.pollingBase, runtime.carrierStore, runtime.carrierSettings, alerts, + steps.Realname, steps.Carddata, steps.CardStatus, steps.Package, carrierCfg, instanceID) + c := &carrierSchedulerRuntime{ + cfg: carrierCfg, runtime: runtime, singleton: singleton, instanceID: instanceID, logger: appLogger, + alerts: alerts, executor: executor, legacyRestarts: legacyRestart, + initializer: polling.NewCarrierInitializer(runtime.carrierScheduling, runtime.pollingIotCardStore, appLogger), + } + if carrierCfg.NSQDAddress != "" { + producer, err := nsqbus.NewProducer(carrierCfg.NSQDAddress, appLogger) + if err != nil { + appLogger.Error("创建 NSQ 生产者失败，按运营商账户调度不可用", zap.Error(err)) + } else { + c.producer = producer + } + c.consumers = nsqbus.NewConsumerManager(nsqbus.ConsumerConfig{ + NSQDAddress: carrierCfg.NSQDAddress, NSQLookupdAddresses: nsqbus.SplitAddresses(carrierCfg.NSQLookupdAddresses), + MsgTimeout: carrierCfg.MsgTimeout, MaxAttempts: carrierCfg.MaxAttempts, + }, constants.NSQChannelWorker, executor.Handle, appLogger) + } + var budgets carrierqueue.BudgetProvider + if runtime.gatewayLimit != nil && runtime.gatewayLimit.Limiter != nil { + budgets = runtime.gatewayLimit.Limiter + } + if c.producer != nil { + c.dispatcher = carrierqueue.NewDispatcher(runtime.db, runtime.carrierStore, runtime.carrierSettings, c.producer, budgets, + carrierCfg, instanceID, appLogger) + if singleton { + go c.dispatcher.Run(ctx) + } + } + go c.watch(ctx) + return c +} + +// RequestReinit 轮询配置变更时请求按新配置重新初始化新结构（仅 carrier 模式生效）。 +func (c *carrierSchedulerRuntime) RequestReinit() { + if c == nil { + return + } + c.mu.Lock() + c.reinitQueued = true + c.mu.Unlock() +} + +func (c *carrierSchedulerRuntime) watch(ctx context.Context) { + ticker := time.NewTicker(3 * time.Second) + defer ticker.Stop() + for { + c.step(ctx) + select { + case <-ctx.Done(): + c.shutdown() + return + case <-ticker.C: + } + } +} + +func (c *carrierSchedulerRuntime) step(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + c.logger.Error("按运营商账户调度模式监听 panic，已恢复", zap.Any("panic", r)) + } + }() + mode := c.runtime.carrierSettings.Current(ctx).Mode + c.mu.Lock() + previous := c.lastMode + c.lastMode = mode + reinit := c.reinitQueued + c.reinitQueued = false + c.mu.Unlock() + if mode == constants.PollingSchedulerModeCarrier { + if previous != mode { + c.logger.Info("卡轮询调度模式：按运营商账户调度", zap.String("previous", previous)) + } + if c.singleton { + c.ensureInitialized(ctx, reinit) + } + c.syncConsumers(ctx) + return + } + if previous == constants.PollingSchedulerModeCarrier { + c.logger.Info("卡轮询调度模式回到旧分片调度：停止领取与消费，重建旧分片队列") + if c.consumers != nil { + c.consumers.StopAll(c.grace()) + c.consumers.Reset() + } + if c.singleton { + c.runtime.redisClient.Del(ctx, constants.RedisCarrierSchedulerInitStateKey) + if c.legacyRestarts != nil { + c.legacyRestarts() + } + } + } +} + +// ensureInitialized 切到 carrier 后初始化新结构；初始化完成前分发器不领取。多实例时用初始化锁协调。 +func (c *carrierSchedulerRuntime) ensureInitialized(ctx context.Context, reinit bool) { + redisClient := c.runtime.redisClient + state, _ := redisClient.Get(ctx, constants.RedisCarrierSchedulerInitStateKey).Result() + if state == carrierqueue.InitStateDone && !reinit { + return + } + ok, err := redisClient.SetNX(ctx, carrierInitLockKey, c.instanceID, 30*time.Minute).Result() + if err != nil || !ok { + return + } + defer redisClient.Del(context.WithoutCancel(ctx), carrierInitLockKey) + if state != carrierqueue.InitStateDone { + redisClient.Set(ctx, constants.RedisCarrierSchedulerInitStateKey, carrierqueue.InitStateRunning, 0) + } + started := time.Now() + written, initErr := c.initializer.Run(ctx) + if initErr != nil { + c.logger.Error("按运营商账户调度初始化失败，下一轮重试", zap.Error(initErr)) + return + } + redisClient.Set(ctx, constants.RedisCarrierSchedulerInitStateKey, carrierqueue.InitStateDone, 0) + c.logger.Info("按运营商账户调度初始化完成，开始领取", zap.Int64("cards", written), zap.Duration("duration", time.Since(started)), + zap.String("metric", "polling.carrier.initialized")) +} + +// syncConsumers 按当前运营商账户订阅卡任务 topic，max_in_flight 取账户调度参数。 +func (c *carrierSchedulerRuntime) syncConsumers(ctx context.Context) { + if c.consumers == nil || c.dispatcher == nil { + return + } + desired := map[string]int{} + for _, carrierID := range c.dispatcher.CarrierIDs(ctx) { + desired[constants.NSQCardTopic(carrierID)] = c.dispatcher.ParamsFor(carrierID).CardMaxInFlight + } + c.consumers.Sync(desired) +} + +func (c *carrierSchedulerRuntime) grace() time.Duration { + if c.cfg.ShutdownGrace > 0 { + return c.cfg.ShutdownGrace + } + return 30 * time.Second +} + +// shutdown 优雅退出：停止订阅，宽限期内等待执行中的卡任务完成；未完成的由租约到期后重新领取。 +func (c *carrierSchedulerRuntime) shutdown() { + if c.consumers != nil { + c.consumers.StopAll(c.grace()) + } + if c.producer != nil { + c.producer.Stop() + } +} + +// InitProgress 返回新结构初始化进度。 +func (c *carrierSchedulerRuntime) InitProgress() polling.InitProgress { + return c.initializer.Progress() +} diff --git a/cmd/worker/main.go b/cmd/worker/main.go index e79a8c7d..02c5f5e6 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -45,6 +45,7 @@ import ( wecomInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/wecom" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/internal/polling" + "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card" pollingSvc "github.com/break/junhong_cmp_fiber/internal/service/polling" refundSvc "github.com/break/junhong_cmp_fiber/internal/service/refund" @@ -102,6 +103,12 @@ type workerRuntime struct { assetAutoRenewalService *assetAutoRenewalApp.Service // gatewayLimit 是卡网关运营商账户限流组件（限流、统计、停复机记录）。 gatewayLimit *gatewaylimit.Runtime + // 按运营商账户调度：调度参数、账户队列存储与模式接缝（模式为 carrier 时各入口写新结构）。 + carrierSettings *carrierqueue.SettingsProvider + carrierStore *carrierqueue.Store + carrierScheduling *polling.CarrierScheduling + // carrierReinit 在 carrier 模式下轮询配置变更时请求重新初始化新结构。 + carrierReinit func() } func main() { @@ -171,6 +178,12 @@ func runWorker(cfg *config.Config) { taskHandler := createTaskHandler(runtime, appLogger) taskHandler.RegisterHandlers() + carrierRuntime := startCarrierScheduler(ctx, cfg, runtime, taskHandler.PollingSteps(), runsSingletonModules(cfg.Worker.Role), func() { + if pollingInitializer != nil { + pollingInitializer.Restart(ctx) + } + }, appLogger) + runtime.carrierReinit = carrierRuntime.RequestReinit registerWeComApprovalTasks(taskHandler.GetMux(), runtime, cfg, appLogger) registerAgentRechargeRecoveryTask(taskHandler.GetMux(), runtime, appLogger) registerRefundChannelRecoveryTask(taskHandler.GetMux(), runtime, appLogger) @@ -379,6 +392,7 @@ func initWorkerRuntime(ctx context.Context, cfg *config.Config, appLogger *zap.L // 自动续费的复机执行同样复用既有停复机单一事实源。 assetAutoRenewalService: workerResult.Services.AssetAutoRenewal, } + newCarrierSchedulerParts(runtime, appLogger) registerNotificationOutboxConsumer(runtime, appLogger) registerWalletOutboxConsumer(runtime, appLogger) registerCardObservationOutboxConsumer(runtime, appLogger) @@ -842,6 +856,13 @@ func startPollingInitializer(ctx context.Context, runtime *workerRuntime, appLog pollingInitializer.StartBackground(ctx) runtime.pollingConfigMgr.WatchChanges(ctx, func(hadConfigs, hasConfigs bool) { + if runtime.carrierScheduling.Active(ctx) { + // 按运营商账户调度时按新配置重新初始化新结构，旧分片队列在回退时再重建。 + if runtime.carrierReinit != nil { + runtime.carrierReinit() + } + return + } if hasConfigs { appLogger.Info("轮询配置已变更，触发队列重新初始化", zap.Bool("had_configs", hadConfigs)) @@ -894,6 +915,7 @@ func startPollingScheduler( dataResetHandler, ) pollingScheduler.SetInitializer(pollingInitializer) + pollingScheduler.SetCarrierScheduling(runtime.carrierScheduling) if err := pollingScheduler.Start(ctx); err != nil { appLogger.Error("启动轮询调度器失败", zap.Error(err)) diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index 4db09d8c..827d33f5 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -29,6 +29,7 @@ import ( walletinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/wallet" wecomInfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/wecom" "github.com/break/junhong_cmp_fiber/internal/polling" + "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" assetAutoRenewalQuery "github.com/break/junhong_cmp_fiber/internal/query/assetautorenewal" accountSvc "github.com/break/junhong_cmp_fiber/internal/service/account" agentOpenAPISvc "github.com/break/junhong_cmp_fiber/internal/service/agent_open_api" @@ -234,6 +235,10 @@ func initServices(s *stores, deps *Dependencies) *services { pollingConfigMgr.Start(context.Background()) pollingQueueMgr := polling.NewPollingQueueManager(deps.Redis, constants.PollingShardCount, deps.Logger) pollingLifecycleSvc := polling.NewPollingLifecycleService(pollingQueueMgr, pollingConfigMgr, s.IotCard, s.DeviceSimBinding, s.Device, deps.Logger) + // 按运营商账户调度：模式为 carrier 时 API 侧生命周期事件写入账户队列与卡到期信息。 + carrierSettings := carrierqueue.NewDefaultSettingsProvider(deps.DB, deps.Redis, deps.Logger) + carrierStore := carrierqueue.NewStore(deps.Redis) + pollingLifecycleSvc.SetCarrierScheduling(polling.NewCarrierScheduling(carrierStore, carrierSettings, pollingConfigMgr, deps.Logger)) iotCard.SetPollingCallback(pollingLifecycleSvc) // 创建支付配置服务（Order 和 Recharge 依赖） wechatConfig := wechatConfigSvc.New(s.WechatConfig, s.Order, s.RechargeOrder, s.AgentRecharge, s.Payment, auditWriter, deps.Redis, deps.Logger) diff --git a/internal/gateway/call_guard.go b/internal/gateway/call_guard.go index 014e8d41..903b6641 100644 --- a/internal/gateway/call_guard.go +++ b/internal/gateway/call_guard.go @@ -91,6 +91,7 @@ type CallGuard interface { type cardCallKey struct{} type carrierKey struct{} type exemptKey struct{} +type rateLimitWaitKey struct{} type cardCall struct { operation string @@ -113,6 +114,21 @@ func WithRateLimitExempt(ctx context.Context, source string) context.Context { return context.WithValue(ctx, exemptKey{}, source) } +// WithRateLimitWait 声明本次调用在拿不到限流名额时等待，总等待不超过 maxWait；超时返回限流错误。 +// 只用于不占 Asynq 协程的执行方（按运营商账户调度的 NSQ 消费者）；未声明时拿不到名额立即返回。 +func WithRateLimitWait(ctx context.Context, maxWait time.Duration) context.Context { + if maxWait <= 0 { + return ctx + } + return context.WithValue(ctx, rateLimitWaitKey{}, maxWait) +} + +// RateLimitWait 返回 ctx 中声明的限流最长等待；未声明返回 0。 +func RateLimitWait(ctx context.Context) time.Duration { + wait, _ := ctx.Value(rateLimitWaitKey{}).(time.Duration) + return wait +} + // RateLimitExemptSource 返回 ctx 中声明的豁免来源，未声明返回空串。 func RateLimitExemptSource(ctx context.Context) string { source, _ := ctx.Value(exemptKey{}).(string) diff --git a/internal/infrastructure/gatewaylimit/guard.go b/internal/infrastructure/gatewaylimit/guard.go index 512e8bfd..cd1c91ad 100644 --- a/internal/infrastructure/gatewaylimit/guard.go +++ b/internal/infrastructure/gatewaylimit/guard.go @@ -22,6 +22,9 @@ const ( statPrefixDuration = "duration_le:" ) +// StatFieldRateLimitWait 是等待模式下进入限流等待的次数字段（等待超时另计入 rate_limited）。 +const StatFieldRateLimitWait = "rate_wait" + // durationBuckets 是耗时分桶上界（毫秒），用于近似计算 p95。 var durationBuckets = []int64{100, 300, 500, 1000, 3000, 10000, 60000} @@ -60,6 +63,18 @@ func (g *Guard) BeforeAttempt(ctx context.Context, call gateway.CallInfo) error g.incr(ctx, carrierID, call.Operation, statFieldWouldLimit) } if !decision.Allowed { + // 等待模式（NSQ 消费者）：等名额直到超时；超时按限流错误返回，调用方按短延迟重排、不计失败。 + if maxWait := gateway.RateLimitWait(ctx); maxWait > 0 { + g.incr(ctx, carrierID, call.Operation, StatFieldRateLimitWait) + allowed, waitErr := g.limiter.Wait(ctx, carrierID, call.Operation, maxWait) + if allowed { + return nil + } + if waitErr != nil && ctx.Err() == nil { + g.logger.Warn("网关限流等待失败，本次放行", zap.Uint("carrier_id", carrierID), zap.String("operation", call.Operation), zap.Error(waitErr)) + return nil + } + } g.incr(ctx, carrierID, call.Operation, constants.GatewayErrorRateLimited) return gateway.NewRateLimitedError() } diff --git a/internal/infrastructure/gatewaylimit/limiter.go b/internal/infrastructure/gatewaylimit/limiter.go index 85ad1931..1a9b17af 100644 --- a/internal/infrastructure/gatewaylimit/limiter.go +++ b/internal/infrastructure/gatewaylimit/limiter.go @@ -189,3 +189,50 @@ func windowMember(nowMS int64) string { _, _ = rand.Read(buf) return strconv.FormatInt(nowMS, 10) + "-" + hex.EncodeToString(buf) } + +// Budget 返回运营商账户各查询类操作所属限流组本秒剩余名额，供分发器按预计调用数领取。 +// groups 为限流组 → 本秒剩余；opGroup 为操作 → 限流组。只统计模式下返回 limited=false（不按 QPS 限制领取）。 +// 全局组只在账户组之外再收紧：剩余名额取两者较小值。 +func (l *Limiter) Budget(ctx context.Context, carrierID uint, ops []string) (groups map[string]int, opGroup map[string]string, limited bool) { + if l == nil || l.redis == nil { + return nil, nil, false + } + settings := l.settings.Current(ctx) + if !settings.Enforce() { + return nil, nil, false + } + groups, opGroup = map[string]int{}, map[string]string{} + now := time.Now().UnixMilli() + since := now - constants.GatewayRateLimitWindow.Milliseconds() + for _, op := range ops { + account, accountOK, global, globalOK := l.groups.Resolve(ctx, carrierID, op) + if !accountOK { + account = GroupRule{GroupCode: defaultGroupCode + ":" + op, MaxQPS: settings.DefaultQPS} + } + code := account.GroupCode + opGroup[op] = code + if _, seen := groups[code]; seen { + continue + } + limit := account.MaxQPS + if !account.HighPriority { + limit -= account.ReservedQPS + } + used, err := l.redis.ZCount(ctx, constants.RedisGatewayQPSKey(carrierID, code), strconv.FormatInt(since, 10), "+inf").Result() + if err != nil { + return nil, nil, false + } + remaining := limit - int(used) + if globalOK { + globalUsed, gerr := l.redis.ZCount(ctx, constants.RedisGatewayGlobalQPSKey(global.GroupCode), strconv.FormatInt(since, 10), "+inf").Result() + if gerr == nil && global.MaxQPS-int(globalUsed) < remaining { + remaining = global.MaxQPS - int(globalUsed) + } + } + if remaining < 0 { + remaining = 0 + } + groups[code] = remaining + } + return groups, opGroup, true +} diff --git a/internal/infrastructure/nsqbus/nsqbus.go b/internal/infrastructure/nsqbus/nsqbus.go new file mode 100644 index 00000000..b5530dff --- /dev/null +++ b/internal/infrastructure/nsqbus/nsqbus.go @@ -0,0 +1,215 @@ +// Package nsqbus 封装卡轮询使用的 NSQ 生产者与按 topic 动态订阅的消费者管理器。 +// NSQ 只承担执行分发与消费者并发控制，业务状态与重试全部在 Redis（add-carrier-polling-scheduler）。 +package nsqbus + +import ( + "context" + "strings" + "sync" + "time" + + "github.com/nsqio/go-nsq" + "go.uber.org/zap" +) + +// Producer 是 nsqd 生产者。 +type Producer struct { + producer *nsq.Producer +} + +// NewProducer 连接 nsqd TCP 地址创建生产者。 +func NewProducer(address string, logger *zap.Logger) (*Producer, error) { + cfg := nsq.NewConfig() + producer, err := nsq.NewProducer(address, cfg) + if err != nil { + return nil, err + } + producer.SetLogger(nsqLogger{logger: logger}, nsq.LogLevelWarning) + return &Producer{producer: producer}, nil +} + +// Publish 同步发布一条消息；topic 不存在时由 nsqd 自动创建。 +func (p *Producer) Publish(topic string, body []byte) error { + return p.producer.Publish(topic, body) +} + +// Stop 停止生产者。 +func (p *Producer) Stop() { + if p != nil && p.producer != nil { + p.producer.Stop() + } +} + +// Handler 处理一条消息；返回后由管理器 FIN。消息处理期间可调用 touch 延长 NSQ 消息超时。 +type Handler func(ctx context.Context, body []byte, touch func()) + +// ConsumerConfig 是消费者参数。 +type ConsumerConfig struct { + NSQDAddress string + NSQLookupdAddresses []string + MsgTimeout time.Duration + MaxAttempts uint16 +} + +// ConsumerManager 按 topic 动态订阅与取消订阅；每个 topic 一个消费者，max_in_flight 控制本进程对该 topic 的并发。 +type ConsumerManager struct { + cfg ConsumerConfig + channel string + handler Handler + logger *zap.Logger + + mu sync.Mutex + consumers map[string]*subscription + ctx context.Context + cancel context.CancelFunc + inflight sync.WaitGroup +} + +type subscription struct { + consumer *nsq.Consumer + maxInFlight int +} + +// NewConsumerManager 创建消费者管理器。 +func NewConsumerManager(cfg ConsumerConfig, channel string, handler Handler, logger *zap.Logger) *ConsumerManager { + if logger == nil { + logger = zap.NewNop() + } + ctx, cancel := context.WithCancel(context.Background()) + return &ConsumerManager{cfg: cfg, channel: channel, handler: handler, logger: logger, + consumers: map[string]*subscription{}, ctx: ctx, cancel: cancel} +} + +// Sync 使订阅集合与 desired（topic → max_in_flight）一致：新增订阅、取消多余订阅、调整 max_in_flight。 +func (m *ConsumerManager) Sync(desired map[string]int) { + m.mu.Lock() + defer m.mu.Unlock() + if m.ctx.Err() != nil { + return + } + for topic, sub := range m.consumers { + if _, keep := desired[topic]; !keep { + sub.consumer.Stop() + delete(m.consumers, topic) + m.logger.Info("取消订阅卡任务 topic", zap.String("topic", topic)) + } + } + for topic, maxInFlight := range desired { + if maxInFlight <= 0 { + maxInFlight = 1 + } + if sub, ok := m.consumers[topic]; ok { + if sub.maxInFlight != maxInFlight { + sub.consumer.ChangeMaxInFlight(maxInFlight) + sub.maxInFlight = maxInFlight + } + continue + } + consumer, err := m.subscribe(topic, maxInFlight) + if err != nil { + m.logger.Warn("订阅卡任务 topic 失败，下一轮重试", zap.String("topic", topic), zap.Error(err)) + continue + } + m.consumers[topic] = &subscription{consumer: consumer, maxInFlight: maxInFlight} + m.logger.Info("订阅卡任务 topic", zap.String("topic", topic), zap.Int("max_in_flight", maxInFlight)) + } +} + +func (m *ConsumerManager) subscribe(topic string, maxInFlight int) (*nsq.Consumer, error) { + cfg := nsq.NewConfig() + cfg.MaxInFlight = maxInFlight + if m.cfg.MsgTimeout > 0 { + cfg.MsgTimeout = m.cfg.MsgTimeout + } + if m.cfg.MaxAttempts > 0 { + cfg.MaxAttempts = m.cfg.MaxAttempts + } + consumer, err := nsq.NewConsumer(topic, m.channel, cfg) + if err != nil { + return nil, err + } + consumer.SetLogger(nsqLogger{logger: m.logger}, nsq.LogLevelWarning) + handlerCtx := m.ctx // subscribe 在持锁时调用，捕获当前处理上下文 + consumer.AddConcurrentHandlers(nsq.HandlerFunc(func(message *nsq.Message) error { + message.DisableAutoResponse() + m.inflight.Add(1) + defer m.inflight.Done() + defer message.Finish() + m.handler(handlerCtx, message.Body, message.Touch) + return nil + }), maxInFlight) + if len(m.cfg.NSQLookupdAddresses) > 0 { + err = consumer.ConnectToNSQLookupds(m.cfg.NSQLookupdAddresses) + } else { + err = consumer.ConnectToNSQD(m.cfg.NSQDAddress) + } + if err != nil { + consumer.Stop() + return nil, err + } + return consumer, nil +} + +// Topics 返回当前已订阅的 topic。 +func (m *ConsumerManager) Topics() []string { + m.mu.Lock() + defer m.mu.Unlock() + topics := make([]string, 0, len(m.consumers)) + for topic := range m.consumers { + topics = append(topics, topic) + } + return topics +} + +// StopAll 停止全部订阅；在宽限期内等待执行中的消息完成并 FIN，超过宽限期后取消处理上下文。 +func (m *ConsumerManager) StopAll(grace time.Duration) { + m.mu.Lock() + for topic, sub := range m.consumers { + sub.consumer.Stop() + delete(m.consumers, topic) + } + m.mu.Unlock() + done := make(chan struct{}) + go func() { + m.inflight.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(grace): + m.logger.Warn("优雅退出宽限期内仍有卡任务未完成，取消执行（由租约到期后重新领取）") + } + m.cancel() +} + +// Reset 在 StopAll 后重新允许订阅（切回 carrier 模式时使用）。 +func (m *ConsumerManager) Reset() { + m.mu.Lock() + defer m.mu.Unlock() + if m.ctx.Err() != nil { + m.ctx, m.cancel = context.WithCancel(context.Background()) + } +} + +// SplitAddresses 解析逗号分隔的地址列表。 +func SplitAddresses(raw string) []string { + var result []string + for _, part := range strings.Split(raw, ",") { + if part = strings.TrimSpace(part); part != "" { + result = append(result, part) + } + } + return result +} + +type nsqLogger struct { + logger *zap.Logger +} + +// Output 实现 go-nsq 的日志接口。 +func (l nsqLogger) Output(_ int, s string) error { + if l.logger != nil { + l.logger.Warn("NSQ", zap.String("message", s)) + } + return nil +} diff --git a/internal/polling/carrier_bridge.go b/internal/polling/carrier_bridge.go new file mode 100644 index 00000000..dc8edaa5 --- /dev/null +++ b/internal/polling/carrier_bridge.go @@ -0,0 +1,199 @@ +package polling + +import ( + "context" + "sync" + "sync/atomic" + "time" + + "go.uber.org/zap" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" + "github.com/break/junhong_cmp_fiber/internal/store/postgres" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// CarrierScheduling 是按运营商账户调度在轮询包内的接缝：判断当前模式，并按卡计算各检查类型到期时间写入账户队列。 +// 所有写入轮询结构的入口（初始化、生命周期、优先提示、手动触发）按当前模式写新结构或旧分片结构。 +type CarrierScheduling struct { + store *carrierqueue.Store + settings *carrierqueue.SettingsProvider + configMgr *PollingConfigManager + logger *zap.Logger +} + +// NewCarrierScheduling 创建按运营商账户调度接缝。 +func NewCarrierScheduling(store *carrierqueue.Store, settings *carrierqueue.SettingsProvider, configMgr *PollingConfigManager, logger *zap.Logger) *CarrierScheduling { + if logger == nil { + logger = zap.NewNop() + } + return &CarrierScheduling{store: store, settings: settings, configMgr: configMgr, logger: logger} +} + +// Active 判断当前是否为按运营商账户调度。 +func (c *CarrierScheduling) Active(ctx context.Context) bool { + return c != nil && c.store != nil && c.settings != nil && c.settings.Current(ctx).CarrierMode() +} + +// Store 返回账户队列存储。 +func (c *CarrierScheduling) Store() *carrierqueue.Store { + if c == nil { + return nil + } + return c.store +} + +// Settings 返回调度参数提供者。 +func (c *CarrierScheduling) Settings() *carrierqueue.SettingsProvider { + if c == nil { + return nil + } + return c.settings +} + +// DuesForCard 按卡匹配的轮询配置计算各检查类型下次到期时间。 +// fromLastCheck 为 true 时按各类型上次检查时间计算（初始化）；否则按当前时间加抖动（新卡、恢复轮询、状态变化）。 +// 间隔为 0 或未配置的类型不返回；重置日前加密窗口规则算出的时间更早时以它为准。 +func (c *CarrierScheduling) DuesForCard(ctx context.Context, card *model.IotCard, fromLastCheck bool) map[string]time.Time { + dues := map[string]time.Time{} + if c == nil || c.configMgr == nil || card == nil { + return dues + } + now := time.Now() + for taskType, info := range c.configMgr.MergedTaskIntervals(card) { + checkType := constants.CheckTypeFromTaskType(taskType) + if checkType == "" || info.Interval <= 0 { + continue + } + var due time.Time + if fromLastCheck { + due = calculateNextCheckTime(lastCheckAtByTaskType(card, taskType), info.Interval, now) + if windowDue := c.configMgr.NextCheckAt(ctx, card, taskType, info, now); windowDue.Before(due) && windowDue.After(now) { + due = windowDue + } + } else { + due = calcInitialDelay(info.Interval) + } + dues[checkType] = due + } + return dues +} + +// UpsertCard 把卡写入所属账户队列；fromLastCheck 含义同 DuesForCard。 +func (c *CarrierScheduling) UpsertCard(ctx context.Context, card *model.IotCard, fromLastCheck bool) { + if c == nil || c.store == nil || card == nil { + return + } + if err := c.store.Upsert(ctx, card.ID, card.CarrierID, c.DuesForCard(ctx, card, fromLastCheck)); err != nil { + c.logger.Warn("写入运营商账户轮询队列失败", zap.Uint("card_id", card.ID), zap.Uint("carrier_id", card.CarrierID), zap.Error(err)) + } +} + +// RemoveCard 把卡移出所属账户队列；carrierID 为 0 时按到期信息中记录的账户移出。 +func (c *CarrierScheduling) RemoveCard(ctx context.Context, cardID, carrierID uint) { + if c == nil || c.store == nil { + return + } + if carrierID == 0 { + due, err := c.store.ReadDue(ctx, cardID) + if err != nil || due.CarrierID == 0 { + return + } + carrierID = due.CarrierID + } + if err := c.store.Remove(ctx, cardID, carrierID); err != nil { + c.logger.Warn("移出运营商账户轮询队列失败", zap.Uint("card_id", cardID), zap.Error(err)) + } +} + +// RequestTaskType 把卡的某检查类型（旧任务类型 polling:xxx）设为立即到期放入账户优先队列（优先提示、手动触发）。 +// 卡不在轮询中时不处理。 +func (c *CarrierScheduling) RequestTaskType(ctx context.Context, cardID, carrierID uint, taskType string) { + checkType := constants.CheckTypeFromTaskType(taskType) + if c == nil || c.store == nil || checkType == "" { + return + } + if _, err := c.store.RequestCheck(ctx, carrierqueue.CheckRequest{CardID: cardID, CarrierID: carrierID, CheckTypes: []string{checkType}}); err != nil { + c.logger.Warn("写入运营商账户优先队列失败", zap.Uint("card_id", cardID), zap.String("task_type", taskType), zap.Error(err)) + } +} + +// CarrierInitializer 从数据库按批初始化全部运营商账户队列与卡到期信息；进度与旧初始化器同结构，供 init-progress 展示。 +type CarrierInitializer struct { + scheduling *CarrierScheduling + iotCardStore *postgres.IotCardStore + logger *zap.Logger + + mu sync.RWMutex + progress InitProgress + running atomic.Bool +} + +// NewCarrierInitializer 创建新结构初始化器。 +func NewCarrierInitializer(scheduling *CarrierScheduling, iotCardStore *postgres.IotCardStore, logger *zap.Logger) *CarrierInitializer { + if logger == nil { + logger = zap.NewNop() + } + return &CarrierInitializer{scheduling: scheduling, iotCardStore: iotCardStore, logger: logger} +} + +// Progress 返回初始化进度快照。 +func (i *CarrierInitializer) Progress() InitProgress { + i.mu.RLock() + defer i.mu.RUnlock() + return i.progress +} + +// Run 同步执行一次全量初始化：按批读取启用轮询的卡，计算各类型到期并写入账户队列。返回写入的卡数。 +func (i *CarrierInitializer) Run(ctx context.Context) (int64, error) { + if !i.running.CompareAndSwap(false, true) { + return 0, nil + } + defer i.running.Store(false) + started := time.Now() + total, err := i.iotCardStore.CountForPolling(ctx) + if err != nil { + i.setProgress(func(p *InitProgress) { p.Status, p.ErrorMessage = "failed", err.Error() }) + return 0, err + } + i.setProgress(func(p *InitProgress) { + *p = InitProgress{TotalCards: total, StartTime: started, Status: "running"} + }) + const batchSize = 2000 + var lastID uint + var written int64 + for { + if err := ctx.Err(); err != nil { + i.setProgress(func(p *InitProgress) { p.Status, p.ErrorMessage = "failed", err.Error() }) + return written, err + } + cards, fetchErr := i.iotCardStore.ListForPollingBatch(ctx, lastID, batchSize) + if fetchErr != nil { + i.setProgress(func(p *InitProgress) { p.Status, p.ErrorMessage = "failed", fetchErr.Error() }) + return written, fetchErr + } + if len(cards) == 0 { + break + } + if tier := i.scheduling.configMgr.TierResolver(); tier != nil { + tier.Prefill(ctx, cards) + } + for _, card := range cards { + i.scheduling.UpsertCard(ctx, card, true) + written++ + } + lastID = cards[len(cards)-1].ID + loaded := written + i.setProgress(func(p *InitProgress) { p.LoadedCards, p.LastBatchTime = loaded, time.Now() }) + } + i.setProgress(func(p *InitProgress) { p.Status = "completed" }) + i.logger.Info("运营商账户轮询结构初始化完成", zap.Int64("cards", written), zap.Duration("duration", time.Since(started))) + return written, nil +} + +func (i *CarrierInitializer) setProgress(update func(*InitProgress)) { + i.mu.Lock() + update(&i.progress) + i.mu.Unlock() +} diff --git a/internal/polling/carrierqueue/alerts.go b/internal/polling/carrierqueue/alerts.go new file mode 100644 index 00000000..ef473984 --- /dev/null +++ b/internal/polling/carrierqueue/alerts.go @@ -0,0 +1,177 @@ +package carrierqueue + +import ( + "context" + "strconv" + "time" + + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// FailureEvent 是一次需要告警的检查失败。 +type FailureEvent struct { + CardID uint + ICCID string + CarrierID uint + CheckType string + Category string + Message string + FailCount int + FirstFailed time.Time + At time.Time +} + +type alertOp struct { + failure *FailureEvent + recovered bool + cardID uint + checkType string +} + +// AlertWriter 异步写卡轮询失败告警：同一卡同一检查类型只保留一条未处理告警，后续失败更新计数； +// 首次成功转为已恢复；同账户窗口内新增告警超阈值时产生账户级告警。写入失败只记日志，不影响轮询。 +type AlertWriter struct { + db *gorm.DB + redis *redis.Client + settings *SettingsProvider + logger *zap.Logger + ch chan alertOp +} + +// NewAlertWriter 创建告警写入器；调用 Start 后开始消费。 +func NewAlertWriter(db *gorm.DB, redisClient *redis.Client, settings *SettingsProvider, logger *zap.Logger) *AlertWriter { + if logger == nil { + logger = zap.NewNop() + } + return &AlertWriter{db: db, redis: redisClient, settings: settings, logger: logger, ch: make(chan alertOp, 2048)} +} + +// Start 启动后台写入协程，直到 ctx 结束。 +func (w *AlertWriter) Start(ctx context.Context) { + go func() { + for { + select { + case <-ctx.Done(): + return + case op := <-w.ch: + w.apply(op) + } + } + }() +} + +// ReportFailure 提交一条失败告警（达到阈值后才调用）。 +func (w *AlertWriter) ReportFailure(event FailureEvent) { + w.offer(alertOp{failure: &event}) +} + +// ReportRecovered 提交一次恢复：该卡该检查类型的未处理告警转为已恢复。 +func (w *AlertWriter) ReportRecovered(cardID uint, checkType string) { + w.offer(alertOp{recovered: true, cardID: cardID, checkType: checkType}) +} + +func (w *AlertWriter) offer(op alertOp) { + if w == nil { + return + } + select { + case w.ch <- op: + default: + w.logger.Warn("卡轮询失败告警队列已满，丢弃本次写入", zap.String("metric", "polling.failure_alert.dropped")) + } +} + +func (w *AlertWriter) apply(op alertOp) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if op.recovered { + err := w.db.WithContext(ctx).Exec(`UPDATE tb_polling_failure_alert SET status = ?, recovered_at = NOW(), updated_at = NOW() + WHERE alert_level = ? AND card_id = ? AND check_type = ? AND status = ?`, + constants.PollingAlertStatusRecovered, constants.PollingAlertLevelCard, op.cardID, op.checkType, constants.PollingAlertStatusOpen).Error + if err != nil { + w.logger.Warn("卡轮询告警转为已恢复失败", zap.Uint("card_id", op.cardID), zap.String("check_type", op.checkType), zap.Error(err)) + } + return + } + event := op.failure + var inserted bool + err := w.db.WithContext(ctx).Raw(`INSERT INTO tb_polling_failure_alert + (alert_level, card_id, iccid, carrier_id, check_type, error_category, fail_count, first_failed_at, last_failed_at, last_error, status) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (alert_level, card_id, carrier_id, check_type) WHERE status = 1 + DO UPDATE SET fail_count = EXCLUDED.fail_count, error_category = EXCLUDED.error_category, + last_failed_at = EXCLUDED.last_failed_at, last_error = EXCLUDED.last_error, updated_at = NOW() + RETURNING (xmax = 0)`, + constants.PollingAlertLevelCard, event.CardID, event.ICCID, event.CarrierID, event.CheckType, event.Category, event.FailCount, + nullableTime(event.FirstFailed), event.At, truncate(event.Message, 500), constants.PollingAlertStatusOpen).Row().Scan(&inserted) + if err != nil { + w.logger.Warn("写卡轮询失败告警失败", zap.Uint("card_id", event.CardID), zap.String("check_type", event.CheckType), zap.Error(err)) + return + } + if !inserted { + return + } + w.logger.Error("卡轮询持续失败告警", zap.Uint("card_id", event.CardID), zap.String("iccid", event.ICCID), + zap.Uint("carrier_id", event.CarrierID), zap.String("check_type", event.CheckType), zap.String("category", event.Category), + zap.Int("fail_count", event.FailCount), zap.String("metric", "polling.failure_alert.opened")) + w.checkAccountAlert(ctx, event.CarrierID) +} + +// checkAccountAlert 同账户窗口内新增卡级告警数超过阈值时，产生（或更新）一条账户级告警。 +func (w *AlertWriter) checkAccountAlert(ctx context.Context, carrierID uint) { + settings := w.settings.Current(ctx) + window, limit := settings.AccountAlertWindow, settings.AccountAlertLimit + if window <= 0 || limit <= 0 || w.redis == nil { + return + } + now := time.Now() + key := constants.RedisCarrierAccountAlertKey(carrierID) + member := strconv.FormatInt(now.UnixNano(), 36) + pipe := w.redis.TxPipeline() + pipe.ZAdd(ctx, key, redis.Z{Score: float64(now.UnixMilli()), Member: member}) + pipe.ZRemRangeByScore(ctx, key, "-inf", strconv.FormatInt(now.Add(-window).UnixMilli(), 10)) + count := pipe.ZCard(ctx, key) + pipe.Expire(ctx, key, window*2) + if _, err := pipe.Exec(ctx); err != nil { + return + } + if int(count.Val()) <= limit { + return + } + var inserted bool + err := w.db.WithContext(ctx).Raw(`INSERT INTO tb_polling_failure_alert + (alert_level, card_id, carrier_id, check_type, error_category, fail_count, first_failed_at, last_failed_at, last_error, status) + VALUES (?, 0, ?, ?, '', ?, ?, ?, ?, ?) + ON CONFLICT (alert_level, card_id, carrier_id, check_type) WHERE status = 1 + DO UPDATE SET fail_count = EXCLUDED.fail_count, last_failed_at = EXCLUDED.last_failed_at, updated_at = NOW() + RETURNING (xmax = 0)`, + constants.PollingAlertLevelCarrier, carrierID, constants.PollingAlertCheckTypeAll, count.Val(), now, now, + "窗口内同账户新增失败告警数超过阈值，通常是账户额度、凭证或上游故障，请联系上游处理账户", constants.PollingAlertStatusOpen).Row().Scan(&inserted) + if err != nil { + w.logger.Warn("写账户级轮询告警失败", zap.Uint("carrier_id", carrierID), zap.Error(err)) + return + } + if inserted { + w.logger.Error("运营商账户轮询失败告警数超过阈值", zap.Uint("carrier_id", carrierID), zap.Int64("count", count.Val()), + zap.Duration("window", window), zap.String("metric", "polling.failure_alert.carrier_opened")) + } +} + +func nullableTime(t time.Time) any { + if t.IsZero() { + return nil + } + return t +} + +func truncate(s string, max int) string { + runes := []rune(s) + if len(runes) <= max { + return s + } + return string(runes[:max]) +} diff --git a/internal/polling/carrierqueue/dispatcher.go b/internal/polling/carrierqueue/dispatcher.go new file mode 100644 index 00000000..02695bd8 --- /dev/null +++ b/internal/polling/carrierqueue/dispatcher.go @@ -0,0 +1,326 @@ +package carrierqueue + +import ( + "context" + "encoding/json" + stderrors "errors" + "strconv" + "sync" + "time" + + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + "gorm.io/gorm" + + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/pkg/config" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// CardMessage 是卡任务 NSQ 消息体。 +type CardMessage struct { + CardID uint `json:"card_id"` + CarrierID uint `json:"carrier_id"` + Token string `json:"token"` + DispatchedAt int64 `json:"dispatched_at"` +} + +// Publisher 发布 NSQ 消息。 +type Publisher interface { + Publish(topic string, body []byte) error +} + +// BudgetProvider 返回账户各查询操作所属限流组本秒剩余名额；limited=false 表示不按 QPS 限制领取。 +type BudgetProvider interface { + Budget(ctx context.Context, carrierID uint, ops []string) (groups map[string]int, opGroup map[string]string, limited bool) +} + +// CarrierParams 是一个运营商账户生效的调度参数（已合并全局默认）。 +type CarrierParams struct { + InflightLimit int + CardMaxInFlight int + PriorityRatio float64 + Configured bool // 是否有该账户的配置行 +} + +// carrierEntry 是分发器缓存的账户信息。 +type carrierEntry struct { + id uint + realnameControl bool +} + +// Dispatcher 每个分发间隔对每个运营商账户抢分发锁、领取到期卡并发布到账户 topic。多实例安全。 +type Dispatcher struct { + db *gorm.DB + store *Store + settings *SettingsProvider + publisher Publisher + budgets BudgetProvider + cfg config.CarrierSchedulerConfig + instanceID string + logger *zap.Logger + + mu sync.Mutex + carriers []carrierEntry + params map[uint]model.CarrierPollingParam + loadedAt time.Time + tightened map[uint]time.Time +} + +// NewDispatcher 创建分发器。 +func NewDispatcher(db *gorm.DB, store *Store, settings *SettingsProvider, publisher Publisher, budgets BudgetProvider, + cfg config.CarrierSchedulerConfig, instanceID string, logger *zap.Logger) *Dispatcher { + if logger == nil { + logger = zap.NewNop() + } + return &Dispatcher{db: db, store: store, settings: settings, publisher: publisher, budgets: budgets, cfg: cfg, + instanceID: instanceID, logger: logger, params: map[uint]model.CarrierPollingParam{}, tightened: map[uint]time.Time{}} +} + +// Run 按分发间隔循环，直到 ctx 结束。 +func (d *Dispatcher) Run(ctx context.Context) { + interval := d.cfg.DispatchInterval + if interval <= 0 { + interval = time.Second + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + d.tick(ctx, interval) + } + } +} + +// Ready 判断新调度是否可领取：模式为 carrier 且新结构初始化已完成。 +func (d *Dispatcher) Ready(ctx context.Context) bool { + if !d.settings.Current(ctx).CarrierMode() { + return false + } + state, err := d.store.redis.Get(ctx, constants.RedisCarrierSchedulerInitStateKey).Result() + return err == nil && state == InitStateDone +} + +func (d *Dispatcher) tick(ctx context.Context, interval time.Duration) { + defer func() { + if r := recover(); r != nil { + d.logger.Error("分发器 panic，已恢复", zap.Any("panic", r)) + } + }() + if !d.Ready(ctx) { + return + } + d.store.redis.HSet(ctx, constants.RedisCarrierSchedulerHeartbeatKey, d.instanceID, time.Now().UnixMilli()) + d.drainLegacyHints(ctx) + d.refreshCarriers(ctx) + d.mu.Lock() + carriers := append([]carrierEntry(nil), d.carriers...) + d.mu.Unlock() + lockTTL := interval * 9 / 10 + if lockTTL <= 0 { + lockTTL = 900 * time.Millisecond + } + for _, carrier := range carriers { + if ctx.Err() != nil { + return + } + ok, err := d.store.redis.SetNX(ctx, constants.RedisCarrierDispatchLockKey(carrier.id), d.instanceID, lockTTL).Result() + if err != nil || !ok { + continue + } + d.dispatchCarrier(ctx, carrier) + } +} + +func (d *Dispatcher) dispatchCarrier(ctx context.Context, carrier carrierEntry) { + params := d.ParamsFor(carrier.id) + limit := params.InflightLimit + // 执行上限不得大于订阅进程 max_in_flight 之和，否则领出的消息在 NSQ 中排队到租约过期；不满足时收紧本轮上限并上报。 + processes := d.cfg.SubscriberProcesses + if processes <= 0 { + processes = 1 + } + if capacity := params.CardMaxInFlight * processes; capacity > 0 && limit > capacity { + limit = capacity + d.reportTightened(carrier.id, params.InflightLimit, capacity) + } + input := ClaimInput{ + CarrierID: carrier.id, Lease: d.lease(), InflightLimit: limit, PriorityRatio: params.PriorityRatio, + TokenPrefix: d.instanceID + ":" + strconv.FormatInt(time.Now().UnixNano(), 36), + CountRealname: carrier.realnameControl, + } + if d.budgets != nil { + ops := []string{constants.GatewayOpQueryStatus, constants.GatewayOpQueryFlow, constants.GatewayOpQueryRealname} + if groups, opGroup, limited := d.budgets.Budget(ctx, carrier.id, ops); limited { + input.GroupBudget = groups + input.TypeGroup = map[string]string{ + constants.CheckTypeCardStatus: opGroup[constants.GatewayOpQueryStatus], + constants.CheckTypeCarddata: opGroup[constants.GatewayOpQueryFlow], + constants.CheckTypeRealname: opGroup[constants.GatewayOpQueryRealname], + } + } + } + claimed, expired, err := d.store.Claim(ctx, input) + if err != nil { + d.logger.Warn("运营商账户领取失败", zap.Uint("carrier_id", carrier.id), zap.Error(err)) + return + } + if expired > 0 { + minute := strconv.FormatInt(time.Now().Unix()/60, 10) + key := constants.RedisCarrierExpiredClaimsKey(carrier.id) + d.store.redis.HIncrBy(ctx, key, minute, expired) + d.store.redis.Expire(ctx, key, 2*time.Hour) + d.logger.Warn("运营商账户处理中标记到期（执行崩溃或超时）", zap.Uint("carrier_id", carrier.id), zap.Int64("count", expired), + zap.String("metric", "polling.carrier.claim_expired")) + } + topic := constants.NSQCardTopic(carrier.id) + for _, item := range claimed { + body, _ := json.Marshal(CardMessage{CardID: item.CardID, CarrierID: carrier.id, Token: item.Token, DispatchedAt: time.Now().UnixMilli()}) + if err := d.publisher.Publish(topic, body); err != nil { + // 发布失败：用独立短超时上下文按 token 删除处理中标记并恢复 score；恢复也失败则等租约到期后自然重新领取。 + releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second) + if releaseErr := d.store.Release(releaseCtx, item.CardID, carrier.id, item.Token); releaseErr != nil { + d.logger.Warn("发布失败后释放处理中标记失败", zap.Uint("card_id", item.CardID), zap.Error(releaseErr)) + } + cancel() + d.logger.Warn("发布卡任务失败，已释放", zap.Uint("card_id", item.CardID), zap.String("topic", topic), zap.Error(err)) + } + } +} + +func (d *Dispatcher) lease() time.Duration { + if d.cfg.Lease > 0 { + return d.cfg.Lease + } + return 5 * time.Minute +} + +func (d *Dispatcher) reportTightened(carrierID uint, configured, capacity int) { + d.mu.Lock() + last := d.tightened[carrierID] + if time.Since(last) < 10*time.Minute { + d.mu.Unlock() + return + } + d.tightened[carrierID] = time.Now() + d.mu.Unlock() + d.logger.Warn("账户执行上限大于订阅进程 max_in_flight 之和，已按消费能力收紧本轮领取上限", + zap.Uint("carrier_id", carrierID), zap.Int("inflight_limit", configured), zap.Int("capacity", capacity), + zap.String("metric", "polling.carrier.inflight_tightened")) +} + +// refreshCarriers 每 10 秒重新读取运营商账户与调度参数。 +func (d *Dispatcher) refreshCarriers(ctx context.Context) { + d.mu.Lock() + fresh := time.Since(d.loadedAt) < 10*time.Second + d.mu.Unlock() + if fresh { + return + } + var carriers []model.Carrier + if err := d.db.WithContext(ctx).Select("id", "realname_control_enabled").Order("id").Find(&carriers).Error; err != nil { + d.logger.Warn("读取运营商账户失败，沿用上一次列表", zap.Error(err)) + return + } + var rows []model.CarrierPollingParam + if err := d.db.WithContext(ctx).Find(&rows).Error; err != nil { + d.logger.Warn("读取运营商账户调度参数失败，沿用上一次参数", zap.Error(err)) + return + } + entries := make([]carrierEntry, 0, len(carriers)) + for _, carrier := range carriers { + entries = append(entries, carrierEntry{id: carrier.ID, realnameControl: carrier.RealnameControlEnabled == 1}) + } + params := make(map[uint]model.CarrierPollingParam, len(rows)) + for _, row := range rows { + params[row.CarrierID] = row + } + d.mu.Lock() + d.carriers, d.params, d.loadedAt = entries, params, time.Now() + d.mu.Unlock() +} + +// CarrierIDs 返回当前缓存的运营商账户 ID（消费者据此订阅）。 +func (d *Dispatcher) CarrierIDs(ctx context.Context) []uint { + d.refreshCarriers(ctx) + d.mu.Lock() + defer d.mu.Unlock() + ids := make([]uint, 0, len(d.carriers)) + for _, carrier := range d.carriers { + ids = append(ids, carrier.id) + } + return ids +} + +// ParamsFor 返回账户生效调度参数（未配置项用全局默认）。 +func (d *Dispatcher) ParamsFor(carrierID uint) CarrierParams { + d.mu.Lock() + row, ok := d.params[carrierID] + d.mu.Unlock() + return MergeParams(row, ok, d.cfg) +} + +// MergeParams 合并账户配置行与全局默认。 +func MergeParams(row model.CarrierPollingParam, configured bool, cfg config.CarrierSchedulerConfig) CarrierParams { + params := CarrierParams{InflightLimit: cfg.DefaultInflightLimit, CardMaxInFlight: cfg.DefaultCardMaxInFlight, Configured: configured} + if params.InflightLimit <= 0 { + params.InflightLimit = 20 + } + if params.CardMaxInFlight <= 0 { + params.CardMaxInFlight = 20 + } + if !configured { + return params + } + if row.InflightLimit != nil && *row.InflightLimit > 0 { + params.InflightLimit = *row.InflightLimit + } + if row.CardMaxInFlight != nil && *row.CardMaxInFlight > 0 { + params.CardMaxInFlight = *row.CardMaxInFlight + } + if row.PriorityMaxRatio != nil && *row.PriorityMaxRatio > 0 { + params.PriorityRatio = *row.PriorityMaxRatio + } + return params +} + +// drainLegacyHints 消费旧优先提示通道与手动触发队列：优先提示放入账户优先队列，手动触发在普通队列中立即到期。 +func (d *Dispatcher) drainLegacyHints(ctx context.Context) { + for _, checkType := range constants.AllCheckTypes() { + taskType := constants.TaskTypeFromCheckType(checkType) + d.drainList(ctx, constants.RedisPollingPriorityQueueKey(taskType), checkType, false) + d.drainList(ctx, constants.RedisPollingManualQueueKey(taskType), checkType, true) + } +} + +func (d *Dispatcher) drainList(ctx context.Context, key, checkType string, normal bool) { + for i := 0; i < 500; i++ { + raw, err := d.store.redis.LPop(ctx, key).Result() + if err != nil { + if !stderrors.Is(err, redis.Nil) { + d.logger.Warn("读取旧提示队列失败", zap.String("key", key), zap.Error(err)) + } + return + } + cardID, convErr := strconv.ParseUint(raw, 10, 64) + if convErr != nil { + continue + } + due, dueErr := d.store.ReadDue(ctx, uint(cardID)) + if dueErr != nil || due.CarrierID == 0 { + continue + } + if _, err := d.store.RequestCheck(ctx, CheckRequest{CardID: uint(cardID), CarrierID: due.CarrierID, CheckTypes: []string{checkType}, Normal: normal}); err != nil { + d.logger.Warn("写入账户队列失败", zap.Uint64("card_id", cardID), zap.Error(err)) + } + } +} + +// 新结构初始化状态（constants.RedisCarrierSchedulerInitStateKey）。 +const ( + InitStateRunning = "running" + InitStateDone = "done" +) diff --git a/internal/polling/carrierqueue/scripts.go b/internal/polling/carrierqueue/scripts.go index 621faa65..7b6fb236 100644 --- a/internal/polling/carrierqueue/scripts.go +++ b/internal/polling/carrierqueue/scripts.go @@ -182,7 +182,8 @@ return 1 // 只作用于仍在轮询中的卡（在优先或普通队列中）。冷却期内（最近成功距今未超过冷却）的类型不请求。 // // KEYS[1] due；KEYS[2] 优先队列；KEYS[3] 普通队列；KEYS[4] claim。 -// ARGV：now_ms、card_id、cooldown_ms（0 不冷却）、确认延迟（秒，逗号分隔，可空）、期望结果（可空）、类型…。 +// ARGV：now_ms、card_id、cooldown_ms（0 不冷却）、确认延迟（秒，逗号分隔，可空）、期望结果（可空）、 +// 是否放入优先队列（1/0；0 用于手动触发：只把到期设为现在并在普通队列中立即到期）、类型…。 // 返回本次实际请求的类型数；卡不在轮询中返回 -1。 const requestCheckLua = ` local now = tonumber(ARGV[1]) @@ -192,7 +193,8 @@ 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 toPriority = ARGV[6] == '1' +for i = 7, #ARGV do local t = ARGV[i] local skip = false if cooldown > 0 then @@ -209,6 +211,11 @@ for i = 6, #ARGV do end end if requested == 0 then return 0 end +if not toPriority then + if redis.call('EXISTS', KEYS[4]) == 1 then return requested end + if inQueue then redis.call('ZADD', KEYS[3], now, cardId) else redis.call('ZADD', KEYS[2], now, cardId) end + return requested +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) diff --git a/internal/polling/carrierqueue/settings.go b/internal/polling/carrierqueue/settings.go index 7fc01232..d113307a 100644 --- a/internal/polling/carrierqueue/settings.go +++ b/internal/polling/carrierqueue/settings.go @@ -10,7 +10,9 @@ import ( "sync" "time" + "github.com/redis/go-redis/v9" "go.uber.org/zap" + "gorm.io/gorm" "github.com/break/junhong_cmp_fiber/internal/infrastructure/systemconfig" "github.com/break/junhong_cmp_fiber/pkg/constants" @@ -376,3 +378,15 @@ func (p *SettingsProvider) load(ctx context.Context) (Settings, error) { } return s, nil } + +// NewDefaultSettingsProvider 用独立注册表与读取器创建调度参数提供者（API 与 Worker 共用同一份数据库与缓存数据）。 +func NewDefaultSettingsProvider(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) *SettingsProvider { + registry := systemconfig.NewRegistry() + RegisterConfigDefinitions(registry, logger) + var cache systemconfig.Cache + if redisClient != nil { + cache = systemconfig.NewRedisCache(redisClient) + } + reader := systemconfig.NewReader(db, registry, cache, systemconfig.NewLogAlertSink(logger)) + return NewSettingsProvider(reader, logger) +} diff --git a/internal/polling/carrierqueue/store.go b/internal/polling/carrierqueue/store.go index f4c086e0..8894c333 100644 --- a/internal/polling/carrierqueue/store.go +++ b/internal/polling/carrierqueue/store.go @@ -136,6 +136,8 @@ type CheckRequest struct { Cooldown time.Duration ConfirmDelays []time.Duration Expected string // 延迟确认的期望结果（card_status 为 online / offline），满足时取消后续确认 + // Normal 为 true 时不放入优先队列：只把到期设为现在、在普通队列中立即到期（手动触发走普通通道）。 + Normal bool } // RequestCheck 把卡的指定检查类型设为立即到期放入优先队列；返回实际请求的类型数，卡不在轮询中返回 -1。 @@ -147,7 +149,11 @@ func (s *Store) RequestCheck(ctx context.Context, request CheckRequest) (int64, 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} + toPriority := "1" + if request.Normal { + toPriority = "0" + } + args := []any{time.Now().UnixMilli(), request.CardID, request.Cooldown.Milliseconds(), strings.Join(confirm, ","), request.Expected, toPriority} for _, checkType := range request.CheckTypes { args = append(args, checkType) } diff --git a/internal/polling/lifecycle_service.go b/internal/polling/lifecycle_service.go index dc2d603f..57721006 100644 --- a/internal/polling/lifecycle_service.go +++ b/internal/polling/lifecycle_service.go @@ -22,6 +22,18 @@ type PollingLifecycleService struct { deviceBindingStore *postgres.DeviceSimBindingStore deviceStore *postgres.DeviceStore logger *zap.Logger + // carrier 按运营商账户调度接缝；当前模式为 carrier 时生命周期事件写入账户队列，不写旧分片队列。 + carrier *CarrierScheduling +} + +// SetCarrierScheduling 注入按运营商账户调度接缝。 +func (s *PollingLifecycleService) SetCarrierScheduling(carrier *CarrierScheduling) { + s.carrier = carrier +} + +// carrierActive 判断当前是否按运营商账户调度。 +func (s *PollingLifecycleService) carrierActive(ctx context.Context) bool { + return s.carrier.Active(ctx) } // NewPollingLifecycleService 创建卡生命周期轮询管理服务 @@ -52,6 +64,10 @@ func (s *PollingLifecycleService) OnCardCreated(ctx context.Context, card *model s.logger.Debug("卡禁用轮询，跳过入队", zap.Uint("card_id", card.ID)) return } + if s.carrierActive(ctx) { + s.carrier.UpsertCard(ctx, card, false) + return + } s.enqueueCard(ctx, card) } @@ -64,6 +80,11 @@ func (s *PollingLifecycleService) OnBatchCardsCreated(ctx context.Context, cards // OnCardStatusChanged 卡状态变化后清理缓存、重新匹配配置并更新队列 func (s *PollingLifecycleService) OnCardStatusChanged(ctx context.Context, cardID uint) { + if s.carrierActive(ctx) { + s.queueMgr.InvalidateCardCache(ctx, cardID) + s.carrierRefresh(ctx, cardID) + return + } if err := s.queueMgr.RemoveFromAllQueues(ctx, cardID); err != nil { s.logger.Warn("卡状态变化：从队列移除失败", zap.Uint("card_id", cardID), zap.Error(err)) } @@ -87,6 +108,13 @@ func (s *PollingLifecycleService) OnBatchCardsStatusChanged(ctx context.Context, if len(cardIDs) == 0 { return } + if s.carrierActive(ctx) { + for _, cardID := range cardIDs { + s.queueMgr.InvalidateCardCache(ctx, cardID) + s.carrierRefresh(ctx, cardID) + } + return + } // 单次 pipeline：批量从当前分片队列移除并清理缓存 if err := s.queueMgr.RemoveBatchFromCurrentShardQueues(ctx, cardIDs); err != nil { s.logger.Warn("批量卡状态变化：从队列移除失败", zap.Int("count", len(cardIDs)), zap.Error(err)) @@ -107,6 +135,11 @@ func (s *PollingLifecycleService) OnBatchCardsStatusChanged(ctx context.Context, // OnCardDeleted 卡删除后移除所有队列并清理缓存 func (s *PollingLifecycleService) OnCardDeleted(ctx context.Context, cardID uint) { + if s.carrierActive(ctx) { + s.carrier.RemoveCard(ctx, cardID, 0) + s.queueMgr.InvalidateCardCache(ctx, cardID) + return + } if err := s.queueMgr.OnCardDeleted(ctx, cardID); err != nil { s.logger.Warn("卡删除：清理队列和缓存失败", zap.Uint("card_id", cardID), zap.Error(err)) } @@ -123,11 +156,21 @@ func (s *PollingLifecycleService) OnCardEnabled(ctx context.Context, cardID uint if !s.ShouldEnqueue(ctx, card) { return } + if s.carrierActive(ctx) { + // 恢复轮询时各检查类型到期设为当前时间加抖动。 + s.carrier.UpsertCard(ctx, card, false) + return + } s.enqueueCard(ctx, card) } // OnCardDisabled 卡禁用轮询后移除所有队列 func (s *PollingLifecycleService) OnCardDisabled(ctx context.Context, cardID uint) { + if s.carrierActive(ctx) { + s.carrier.RemoveCard(ctx, cardID, 0) + s.queueMgr.InvalidateCardCache(ctx, cardID) + return + } if err := s.queueMgr.RemoveFromAllQueues(ctx, cardID); err != nil { s.logger.Warn("卡禁用：从队列移除失败", zap.Uint("card_id", cardID), zap.Error(err)) } @@ -183,3 +226,17 @@ func calcInitialDelay(interval int) time.Time { } return now.Add(time.Duration(rand.Intn(jitterMax)) * time.Second) } + +// carrierRefresh 按运营商账户调度时重新匹配配置并写入账户队列：不再轮询的卡移出，其余按当前时间加抖动重排。 +func (s *PollingLifecycleService) carrierRefresh(ctx context.Context, cardID uint) { + card, err := s.iotCardStore.GetByID(ctx, cardID) + if err != nil { + s.logger.Error("卡状态变化：加载卡信息失败", zap.Uint("card_id", cardID), zap.Error(err)) + return + } + if !s.ShouldEnqueue(ctx, card) { + s.carrier.RemoveCard(ctx, card.ID, card.CarrierID) + return + } + s.carrier.UpsertCard(ctx, card, false) +} diff --git a/internal/polling/scheduler.go b/internal/polling/scheduler.go index 483ce7e7..a2bf27bc 100644 --- a/internal/polling/scheduler.go +++ b/internal/polling/scheduler.go @@ -30,6 +30,8 @@ type Scheduler struct { packageActivationHandler *PackageActivationHandler dataResetHandler *DataResetHandler initializer *PollingInitializer // 可选，nil 表示不守卫 Init 完成 + // carrier 按运营商账户调度接缝：当前模式为 carrier 时旧分片调度停止出队（套餐激活检查照常执行）。 + carrier *CarrierScheduling stopChan chan struct{} wg sync.WaitGroup @@ -156,6 +158,10 @@ func (s *Scheduler) scheduleLoop(ctx context.Context) { // processShardSchedule 处理手动队列和分片定时队列（每 1 秒触发） // 使用 90% 的 tick 间隔作为超时，确保单分片 Redis 挂起时不阻塞下一个 tick func (s *Scheduler) processShardSchedule(ctx context.Context) { + if s.carrier.Active(ctx) { + // 已切到按运营商账户调度：旧分片调度停止出队，优先提示与手动触发由新分发器消费。 + return + } // 优先提示与手动队列都不受 Init 影响（ConfigManager 已就绪即可）。 // 同一调度周期内先排空优先提示、再排空手动触发队列：优先的强度边界仅为调度入口领先， // 提示排空不参与分片背压判断，因此分片积压不会抑制加急入队。 @@ -296,3 +302,8 @@ func (s *Scheduler) enqueueBatch(ctx context.Context, taskType string, cardIDs [ } } } + +// SetCarrierScheduling 注入按运营商账户调度接缝。 +func (s *Scheduler) SetCarrierScheduling(carrier *CarrierScheduling) { + s.carrier = carrier +} diff --git a/internal/task/carrier_card_executor.go b/internal/task/carrier_card_executor.go new file mode 100644 index 00000000..b68601bc --- /dev/null +++ b/internal/task/carrier_card_executor.go @@ -0,0 +1,299 @@ +package task + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "sync/atomic" + "time" + + "go.uber.org/zap" + + "github.com/break/junhong_cmp_fiber/internal/gateway" + "github.com/break/junhong_cmp_fiber/internal/model" + "github.com/break/junhong_cmp_fiber/internal/polling" + "github.com/break/junhong_cmp_fiber/internal/polling/carrierqueue" + "github.com/break/junhong_cmp_fiber/pkg/config" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// CarrierCardExecutor 按运营商账户调度的卡任务执行器：一张卡一个卡任务，按配置步骤顺序串行执行已到期的检查类型。 +type CarrierCardExecutor struct { + base *PollingBase + store *carrierqueue.Store + settings *carrierqueue.SettingsProvider + alerts *carrierqueue.AlertWriter + realname *PollingRealnameHandler + carddata *PollingCarddataHandler + cardStatus *PollingCardStatusHandler + pkg *PollingPackageHandler + cfg config.CarrierSchedulerConfig + instanceID string +} + +// NewCarrierCardExecutor 创建卡任务执行器。 +func NewCarrierCardExecutor(base *PollingBase, store *carrierqueue.Store, settings *carrierqueue.SettingsProvider, alerts *carrierqueue.AlertWriter, + realname *PollingRealnameHandler, carddata *PollingCarddataHandler, cardStatus *PollingCardStatusHandler, pkg *PollingPackageHandler, + cfg config.CarrierSchedulerConfig, instanceID string) *CarrierCardExecutor { + return &CarrierCardExecutor{base: base, store: store, settings: settings, alerts: alerts, realname: realname, carddata: carddata, + cardStatus: cardStatus, pkg: pkg, cfg: cfg, instanceID: instanceID} +} + +func (e *CarrierCardExecutor) renewInterval() time.Duration { + if e.cfg.RenewInterval > 0 { + return e.cfg.RenewInterval + } + return 30 * time.Second +} + +func (e *CarrierCardExecutor) lease() time.Duration { + if e.cfg.Lease > 0 { + return e.cfg.Lease + } + return 5 * time.Minute +} + +// Handle 消费一条卡任务消息；返回后由消费者 FIN。重试与退避全部由 Redis score 表达。 +func (e *CarrierCardExecutor) Handle(ctx context.Context, body []byte, touch func()) { + var msg carrierqueue.CardMessage + if err := json.Unmarshal(body, &msg); err != nil || msg.CardID == 0 || msg.Token == "" { + e.base.logger.Warn("卡任务消息无效，丢弃", zap.ByteString("body", body)) + return + } + execValue := msg.Token + "|" + e.instanceID + execTTL := e.renewInterval() * 2 + got, err := e.store.AcquireExec(ctx, msg.CardID, msg.Token, execValue, execTTL) + if err != nil || got != 1 { + // 0：处理中标记已不是该 token（过期或重复投递）；-1：该 token 已有执行者（NSQ 重投时原消费者仍在执行）。 + return + } + settings := e.settings.Current(ctx) + taskCtx, cancel := context.WithTimeout(ctx, settings.TaskTimeout()) + defer cancel() + + var lost atomic.Bool + renewDone := make(chan struct{}) + go func() { + ticker := time.NewTicker(e.renewInterval()) + defer ticker.Stop() + for { + select { + case <-renewDone: + return + case <-taskCtx.Done(): + return + case <-ticker.C: + ok, renewErr := e.store.Renew(context.WithoutCancel(ctx), msg.CardID, msg.CarrierID, msg.Token, execValue, e.lease(), execTTL) + if renewErr == nil && !ok { + // token 已不一致：立即停止后续步骤，不再发起新的网关调用，只写已产生的业务结果。 + lost.Store(true) + cancel() + return + } + if touch != nil { + touch() + } + } + } + }() + set, del := e.execute(taskCtx, msg, settings, &lost) + close(renewDone) + + finishCtx, finishCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer finishCancel() + finished, finishErr := e.store.Finish(finishCtx, msg.CardID, msg.CarrierID, msg.Token, set, del) + if finishErr != nil { + e.base.logger.Warn("卡任务收尾失败，等待租约到期后重新领取", zap.Uint("card_id", msg.CardID), zap.Error(finishErr)) + return + } + if !finished { + e.base.logger.Warn("卡任务收尾时处理中标记已不一致，只写入业务结果", zap.Uint("card_id", msg.CardID), + zap.String("metric", "polling.carrier.finish_token_mismatch")) + } +} + +// execute 串行执行到期检查步骤，返回收尾要写入与删除的到期字段。 +func (e *CarrierCardExecutor) execute(ctx context.Context, msg carrierqueue.CardMessage, settings carrierqueue.Settings, lost *atomic.Bool) (map[string]string, []string) { + set, del := map[string]string{}, []string{} + card, err := e.base.getCardWithCache(ctx, msg.CardID) + if err != nil { + if isNotFound(err) { + e.removeCard(ctx, msg.CardID, msg.CarrierID) + } + return set, del + } + if !card.EnablePolling || (e.base.scopeChecker != nil && !e.base.scopeChecker.ShouldEnqueue(ctx, card)) { + e.removeCard(ctx, msg.CardID, msg.CarrierID) + return set, del + } + due, err := e.store.ReadDue(ctx, msg.CardID) + if err != nil { + return set, del + } + now := time.Now() + flowEvaluated := false + executed := map[string]StepOutcome{} + for _, checkType := range settings.StepOrder { + item, ok := due.Types[checkType] + if !ok || !item.HasDue || item.Due.After(now) { + continue + } + if lost.Load() || ctx.Err() != nil { + break + } + if checkType == constants.CheckTypePackage && settings.SkipPackageAfterFlow && flowEvaluated { + executed[checkType] = stepOK() + continue + } + outcome := e.runStep(ctx, msg, checkType, settings.StepTimeout(checkType)) + if checkType == constants.CheckTypeCarddata && outcome.TriggeredEvaluation { + flowEvaluated = true + } + executed[checkType] = outcome + if outcome.Result == StepCardGone || outcome.Result == StepStopPolling { + e.removeCard(ctx, msg.CardID, msg.CarrierID) + return set, del + } + } + if len(executed) == 0 { + return set, del + } + // 收尾时按最新卡数据重新匹配间隔（步骤中卡状态、套餐可能已变化）。 + e.base.invalidateCardCache(ctx, msg.CardID) + if fresh, freshErr := e.base.getCardWithCache(ctx, msg.CardID); freshErr == nil { + card = fresh + } + intervals := e.base.configMgr.MergedTaskIntervals(card) + now = time.Now() + for checkType, outcome := range executed { + e.writeBack(ctx, card, checkType, outcome, due.Types[checkType], intervals, settings, now, set, &del) + } + return set, del +} + +// runStep 以单步超时、等待模式限流与 panic 恢复执行一个检查步骤。 +func (e *CarrierCardExecutor) runStep(ctx context.Context, msg carrierqueue.CardMessage, checkType string, timeout time.Duration) (outcome StepOutcome) { + stepCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + stepCtx = gateway.WithRateLimitWait(stepCtx, timeout) + defer func() { + if r := recover(); r != nil { + e.base.logger.Error("卡任务步骤 panic，按失败处理", zap.Uint("card_id", msg.CardID), zap.String("check_type", checkType), zap.Any("panic", r)) + outcome = StepOutcome{Result: StepFailed, Category: "panic", Message: fmt.Sprint(r)} + } + }() + switch checkType { + case constants.CheckTypeCardStatus: + return e.cardStatus.Step(stepCtx, msg.CardID) + case constants.CheckTypeCarddata: + return e.carddata.Step(stepCtx, msg.CardID) + case constants.CheckTypeRealname: + return e.realname.Step(stepCtx, msg.CardID) + case constants.CheckTypePackage: + stepCtx = withPollingWorkerAuditContext(stepCtx, constants.TaskTypePollingPackage, "套餐状态轮询任务", msg.Token) + return e.pkg.Step(stepCtx, msg.CardID) + default: + return stepSkip() + } +} + +// writeBack 按步骤结果计算该检查类型下次到期、失败计数与延迟确认，并触发告警或恢复。 +func (e *CarrierCardExecutor) writeBack(ctx context.Context, card *model.IotCard, checkType string, outcome StepOutcome, previous carrierqueue.TypeDue, + intervals map[string]polling.TaskTypeInterval, settings carrierqueue.Settings, now time.Time, set map[string]string, del *[]string) { + info, ok := intervals[constants.TaskTypeFromCheckType(checkType)] + if !ok || info.Interval <= 0 { + // 该类型已不在配置中（间隔为 0 或未配置）：移除到期，不再执行。 + *del = append(*del, carrierqueue.FieldDue(checkType), carrierqueue.FieldFail(checkType), carrierqueue.FieldFirstFail(checkType), + carrierqueue.FieldConfirm(checkType), carrierqueue.FieldExpected(checkType)) + return + } + interval := time.Duration(info.Interval) * time.Second + normalNext := e.base.configMgr.NextCheckAt(ctx, card, constants.TaskTypeFromCheckType(checkType), info, now) + switch outcome.Result { + case StepSucceeded: + next := normalNext + set[carrierqueue.FieldOK(checkType)] = strconv.FormatInt(now.UnixMilli(), 10) + if previous.FailCount > 0 { + e.alerts.ReportRecovered(card.ID, checkType) + } + *del = append(*del, carrierqueue.FieldFail(checkType), carrierqueue.FieldFirstFail(checkType), carrierqueue.FieldError(checkType)) + // 延迟确认：结果已满足预期时取消，否则按序列取下一次确认 = min(正常间隔, 当前 + 确认延迟)。 + if len(previous.Confirm) > 0 { + if previous.Expected != "" && outcome.Observed == previous.Expected { + *del = append(*del, carrierqueue.FieldConfirm(checkType), carrierqueue.FieldExpected(checkType)) + } else { + if confirmAt := now.Add(previous.Confirm[0]); confirmAt.Before(next) { + next = confirmAt + } + if rest := previous.Confirm[1:]; len(rest) > 0 { + set[carrierqueue.FieldConfirm(checkType)] = joinSeconds(rest) + } else { + *del = append(*del, carrierqueue.FieldConfirm(checkType), carrierqueue.FieldExpected(checkType)) + } + } + } + set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(next.UnixMilli(), 10) + case StepFailed: + set[carrierqueue.FieldError(checkType)] = truncateRunes(outcome.Category+"|"+outcome.Message, 300) + if !outcome.CountsAsFailure() { + // 上游频率限制、已是目标状态：不计入连续失败，按首个退避间隔重试。 + set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(settings.BackoffFor(checkType, 1, interval)).UnixMilli(), 10) + return + } + failCount := previous.FailCount + 1 + firstFailed := previous.FirstFailed + if firstFailed.IsZero() { + firstFailed = now + set[carrierqueue.FieldFirstFail(checkType)] = strconv.FormatInt(now.UnixMilli(), 10) + } + set[carrierqueue.FieldFail(checkType)] = strconv.Itoa(failCount) + set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(settings.BackoffFor(checkType, failCount, interval)).UnixMilli(), 10) + if failCount >= settings.AlertThreshold(checkType) || now.Sub(firstFailed) >= settings.AlertFailDuration || settings.ImmediateAlert[outcome.Category] { + e.alerts.ReportFailure(carrierqueue.FailureEvent{CardID: card.ID, ICCID: card.ICCID, CarrierID: card.CarrierID, CheckType: checkType, + Category: outcome.Category, Message: outcome.Message, FailCount: failCount, FirstFailed: firstFailed, At: now}) + } + case StepRateLimited: + // 限流等待超时：请求未发出，按短延迟重排，不计失败。 + delay := 5 * time.Second + if e.base.rateLimitedDelay != nil { + if configured := e.base.rateLimitedDelay(); configured > 0 { + delay = configured + } + } + set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(delay).UnixMilli(), 10) + case StepDeferred: + delay := outcome.DeferDelay + if delay <= 0 { + delay = 30 * time.Second + } + set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(now.Add(delay).UnixMilli(), 10) + default: + set[carrierqueue.FieldDue(checkType)] = strconv.FormatInt(normalNext.UnixMilli(), 10) + } +} + +func (e *CarrierCardExecutor) removeCard(ctx context.Context, cardID, carrierID uint) { + removeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second) + defer cancel() + if err := e.store.Remove(removeCtx, cardID, carrierID); err != nil { + e.base.logger.Warn("移出运营商账户轮询队列失败", zap.Uint("card_id", cardID), zap.Error(err)) + } +} + +func joinSeconds(delays []time.Duration) string { + parts := make([]string, 0, len(delays)) + for _, delay := range delays { + parts = append(parts, strconv.FormatInt(int64(delay/time.Second), 10)) + } + return strings.Join(parts, ",") +} + +func truncateRunes(s string, max int) string { + runes := []rune(s) + if len(runes) <= max { + return s + } + return string(runes[:max]) +} diff --git a/internal/task/polling_base.go b/internal/task/polling_base.go index 064f5ecb..4e347f7e 100644 --- a/internal/task/polling_base.go +++ b/internal/task/polling_base.go @@ -71,6 +71,8 @@ type PollingBase struct { priorityAudit *audit.Writer scopeChecker *polling.PollingLifecycleService rateLimitedDelay func() time.Duration // 卡网关调用被运营商账户限流后的固定重入队延迟 + errorClassifier ErrorClassifier // 上游错误分类（按运营商账户调度的失败退避与告警使用） + schedulerMode SchedulerModeReader // 当前调度模式；carrier 时旧 Asynq 轮询任务直接结束 } // SetRateLimitedDelay 设置被运营商账户限流后的重入队延迟来源（受控系统配置，即时生效）。 diff --git a/internal/task/polling_carddata_handler.go b/internal/task/polling_carddata_handler.go index d4d330c4..22b077fe 100644 --- a/internal/task/polling_carddata_handler.go +++ b/internal/task/polling_carddata_handler.go @@ -30,34 +30,42 @@ func NewPollingCarddataHandler(base *PollingBase, gw *gateway.Client, carrier *p return &PollingCarddataHandler{base: base, gateway: gw, carrier: carrier, observation: observation, integration: integration} } -// Handle 处理流量检查任务。 +// Handle 处理流量检查任务（旧分片调度）。按运营商账户调度时直接结束。 func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) error { - startedAt := time.Now() cardID, ok := parseTaskPayload(task.Payload(), h.base.logger) if !ok { return nil } + if h.base.carrierSchedulerActive(ctx) { + return nil + } if !h.base.acquireConcurrency(ctx, constants.TaskTypePollingCarddata) { h.base.logger.Debug("并发已满，重新入队", zap.Uint("card_id", cardID)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) } defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingCarddata) + return h.base.applyLegacyOutcome(ctx, cardID, constants.TaskTypePollingCarddata, h.Step(ctx, cardID)) +} + +// Step 执行一次流量检查步骤，返回结果分类；不负责重入队。旧分片调度与按运营商账户调度共用。 +func (h *PollingCarddataHandler) Step(ctx context.Context, cardID uint) StepOutcome { + startedAt := time.Now() lockToken, locked, err := h.base.acquireCardTrafficSyncLock(ctx, cardID) if err != nil || !locked { h.base.logger.Warn("卡流量同步锁不可用，延迟重入队", zap.Uint("card_id", cardID), zap.Error(err)) - return h.base.queueMgr.Requeue(ctx, cardID, constants.TaskTypePollingCarddata, time.Now().Add(5*time.Second)) + return stepDefer(5 * time.Second) } defer h.base.releaseCardTrafficSyncLock(ctx, cardID, lockToken) // 优先轮询互斥接缝：放在流量同步锁获取成功之后，避免认领后拿不到锁导致优先项空占。 priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingCarddata) if priorityErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr) + return h.fail(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr) } if skip { h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后", zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingCarddata)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) + return stepDefer(0) } h.base.invalidateCardCache(ctx, cardID) @@ -65,42 +73,42 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e if err != nil { if isNotFound(err) { priority.FailFinal(ctx, "卡不存在") - return nil + return StepOutcome{Result: StepCardGone} } // 本地读卡失败属可恢复失败：先按真实执行累加尝试次数，再按上限收敛，避免加急事实被静默丢弃。 priority.MarkExecution(ctx) priority.FailRetryable(ctx, "获取卡信息失败") - return h.failAndRequeue(ctx, cardID, startedAt, "获取卡信息失败", err) + return h.fail(ctx, cardID, startedAt, "获取卡信息失败", err) } // 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。 if !h.base.priorityInPollingScope(ctx, priority, card) { - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) + return stepSkip() } if h.gateway == nil { priority.FailFinal(ctx, "流量查询能力未配置") - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) + return stepSkip() } priority.MarkExecution(ctx) attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayTraffic, constants.CardObservationSceneTrafficPolling) result, err := h.gateway.QueryFlow(gateway.WithCarrier(ctx, card.CarrierID), &gateway.FlowQueryReq{CardNo: card.ICCID}) if gateway.IsRateLimited(err) { priority.FailRetryable(ctx, "运营商账户网关限流，稍后重试") - return h.base.requeueRateLimited(cardID, constants.TaskTypePollingCarddata) + return StepOutcome{Result: StepRateLimited} } if err != nil { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询流量失败", priority) priority.FailRetryable(ctx, "查询流量失败") - return h.failAndRequeue(ctx, cardID, startedAt, "查询流量失败", err) + return h.fail(ctx, cardID, startedAt, "查询流量失败", err) } if strings.TrimSpace(result.ICCID) == "" { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "流量查询响应缺少 ICCID", priority) priority.FailRetryable(ctx, "流量查询响应缺少 ICCID") - return h.failAndRequeue(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil) + return h.fail(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil) } ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingCarddata, "卡流量轮询任务", attempt.IntegrationID) if h.observation == nil || h.carrier == nil { priority.FailFinal(ctx, "卡流量观测能力未配置") - return h.failAndRequeue(ctx, cardID, startedAt, "卡流量观测能力未配置", nil) + return h.fail(ctx, cardID, startedAt, "卡流量观测能力未配置", nil) } now := time.Now() decision, err := h.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{ @@ -113,7 +121,7 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e if err != nil { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用流量观测失败", priority) priority.FailRetryable(ctx, "应用流量观测失败") - return h.failAndRequeue(ctx, cardID, startedAt, "应用流量观测失败", err) + return h.fail(ctx, cardID, startedAt, "应用流量观测失败", err) } h.base.configMgr.RecordPreResetCheck(ctx, card, now) changed := decision.IncrementMB > 0 || decision.CrossMonth @@ -132,15 +140,15 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e h.base.logger.Info("流量轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric)) priority.Complete(ctx) h.base.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startedAt)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) + return StepOutcome{Result: StepSucceeded, TriggeredEvaluation: changed} } -func (h *PollingCarddataHandler) failAndRequeue(ctx context.Context, cardID uint, startedAt time.Time, message string, err error) error { +func (h *PollingCarddataHandler) fail(ctx context.Context, cardID uint, startedAt time.Time, message string, err error) StepOutcome { fields := []zap.Field{zap.Uint("card_id", cardID)} if err != nil { fields = append(fields, zap.Error(err)) } h.base.logger.Warn(message, fields...) h.base.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startedAt)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata) + return h.base.stepFail(ctx, constants.GatewayOpQueryFlow, message, err) } diff --git a/internal/task/polling_cardstatus_handler.go b/internal/task/polling_cardstatus_handler.go index 9a3bafe4..1a1acdc0 100644 --- a/internal/task/polling_cardstatus_handler.go +++ b/internal/task/polling_cardstatus_handler.go @@ -37,70 +37,85 @@ func NewPollingCardStatusHandler(base *PollingBase, gw *gateway.Client, observat return &PollingCardStatusHandler{base: base, gateway: gw, observation: observation, integration: integration} } -// Handle 处理卡状态轮询任务。 +// Handle 处理卡状态轮询任务（旧分片调度）。按运营商账户调度时直接结束。 func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task) error { - startedAt := time.Now() cardID, ok := parseTaskPayload(task.Payload(), h.base.logger) if !ok { return nil } + if h.base.carrierSchedulerActive(ctx) { + return nil + } if !h.base.acquireConcurrency(ctx, constants.TaskTypePollingCardStatus) { h.base.logger.Debug("并发已满，重新入队", zap.Uint("card_id", cardID)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) } defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingCardStatus) + outcome := h.Step(ctx, cardID) + if outcome.Result == StepSucceeded { + // 卡状态任务仍正常但套餐任务丢失时，仅补入缺失项；不改写已有套餐任务的执行时间。 + h.base.invalidateCardCache(ctx, cardID) + if err := h.base.ensureMissingTask(ctx, cardID, constants.TaskTypePollingPackage); err != nil { + h.base.logger.Warn("卡状态轮询补齐套餐任务失败", zap.Uint("card_id", cardID), zap.Error(err)) + } + } + return h.base.applyLegacyOutcome(ctx, cardID, constants.TaskTypePollingCardStatus, outcome) +} +// Step 执行一次卡状态检查步骤，返回结果分类与观测到的网络状态；不负责重入队。旧分片调度与按运营商账户调度共用。 +func (h *PollingCardStatusHandler) Step(ctx context.Context, cardID uint) StepOutcome { + startedAt := time.Now() // 优先轮询互斥接缝：存在活动优先项时本轮即优先执行或被跳过；无活动项时行为与既有实现一致。 priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingCardStatus) if priorityErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr) + return h.fail(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr) } if skip { h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后", zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingCardStatus)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) + return stepDefer(0) } card, err := h.base.getCardWithCache(ctx, cardID) if err != nil { if isNotFound(err) { priority.FailFinal(ctx, "卡不存在") - return nil + return StepOutcome{Result: StepCardGone} } // 本地读卡失败属可恢复失败：先按真实执行累加尝试次数，再按上限收敛，避免加急事实被静默丢弃。 priority.MarkExecution(ctx) priority.FailRetryable(ctx, "获取卡信息失败") - return h.failAndRequeue(ctx, cardID, startedAt, "获取卡信息失败", err) + return h.fail(ctx, cardID, startedAt, "获取卡信息失败", err) } // 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。 if !h.base.priorityInPollingScope(ctx, priority, card) { - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) + return stepSkip() } if h.gateway == nil { priority.FailFinal(ctx, "卡状态查询能力未配置") - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) + return stepSkip() } priority.MarkExecution(ctx) attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayNetwork, constants.CardObservationSceneNetworkPolling) result, err := h.gateway.QueryCardStatus(gateway.WithCarrier(ctx, card.CarrierID), &gateway.CardStatusReq{CardNo: card.ICCID}) if gateway.IsRateLimited(err) { priority.FailRetryable(ctx, "运营商账户网关限流，稍后重试") - return h.base.requeueRateLimited(cardID, constants.TaskTypePollingCardStatus) + return StepOutcome{Result: StepRateLimited} } if err != nil { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询卡状态失败", priority) priority.FailRetryable(ctx, "查询卡状态失败") - return h.failAndRequeue(ctx, cardID, startedAt, "查询卡状态失败", err) + return h.fail(ctx, cardID, startedAt, "查询卡状态失败", err) } if strings.TrimSpace(result.ICCID) == "" { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "卡状态查询响应缺少 ICCID", priority) priority.FailRetryable(ctx, "卡状态查询响应缺少 ICCID") - return h.failAndRequeue(ctx, cardID, startedAt, "卡状态查询响应缺少 ICCID", nil) + return h.fail(ctx, cardID, startedAt, "卡状态查询响应缺少 ICCID", nil) } ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingCardStatus, "卡网络状态轮询任务", attempt.IntegrationID) if h.observation == nil { priority.FailFinal(ctx, "卡网络观测能力未配置") - return h.failAndRequeue(ctx, cardID, startedAt, "卡网络观测能力未配置", nil) + return h.fail(ctx, cardID, startedAt, "卡网络观测能力未配置", nil) } now := time.Now() decision, err := h.observation.ApplyNetworkObservation(ctx, carddomain.NetworkObservation{ @@ -113,7 +128,7 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task) if err != nil { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用网络状态观测失败", priority) priority.FailRetryable(ctx, "应用网络状态观测失败") - return h.failAndRequeue(ctx, cardID, startedAt, "应用网络状态观测失败", err) + return h.fail(ctx, cardID, startedAt, "应用网络状态观测失败", err) } changed := decision.StatusChanged || decision.StopReasonChanged || decision.StopPolling if changed { @@ -134,23 +149,24 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task) h.base.updateStats(ctx, constants.TaskTypePollingCardStatus, true, time.Since(startedAt)) if decision.StopPolling { h.base.logger.Info("卡命中风险停机或已销户，已关闭该卡轮询", zap.Uint("card_id", cardID), zap.String("gateway_extend", decision.GatewayExtend)) - return nil + return StepOutcome{Result: StepStopPolling} } - - // 卡状态任务仍正常但套餐任务丢失时，仅补入缺失项；不改写已有套餐任务的执行时间。 - h.base.invalidateCardCache(ctx, cardID) - if err := h.base.ensureMissingTask(ctx, cardID, constants.TaskTypePollingPackage); err != nil { - h.base.logger.Warn("卡状态轮询补齐套餐任务失败", zap.Uint("card_id", cardID), zap.Error(err)) + outcome := stepOK() + if decision.StatusKnown { + outcome.Observed = ObservedOffline + if decision.AfterStatus == constants.NetworkStatusOnline { + outcome.Observed = ObservedOnline + } } - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) + return outcome } -func (h *PollingCardStatusHandler) failAndRequeue(ctx context.Context, cardID uint, startedAt time.Time, message string, err error) error { +func (h *PollingCardStatusHandler) fail(ctx context.Context, cardID uint, startedAt time.Time, message string, err error) StepOutcome { fields := []zap.Field{zap.Uint("card_id", cardID)} if err != nil { fields = append(fields, zap.Error(err)) } h.base.logger.Warn(message, fields...) h.base.updateStats(ctx, constants.TaskTypePollingCardStatus, false, time.Since(startedAt)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus) + return h.base.stepFail(ctx, constants.GatewayOpQueryStatus, message, err) } diff --git a/internal/task/polling_package_handler.go b/internal/task/polling_package_handler.go index 00d67e01..15d7d520 100644 --- a/internal/task/polling_package_handler.go +++ b/internal/task/polling_package_handler.go @@ -37,14 +37,15 @@ func NewPollingPackageHandler( } } -// Handle 处理套餐检查任务 +// Handle 处理套餐检查任务（旧分片调度）。按运营商账户调度时直接结束。 func (h *PollingPackageHandler) Handle(ctx context.Context, t *asynq.Task) error { - startTime := time.Now() - cardID, ok := parseTaskPayload(t.Payload(), h.base.logger) if !ok { return nil } + if h.base.carrierSchedulerActive(ctx) { + return nil + } ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingPackage, "套餐状态轮询任务", t.ResultWriter().TaskID()) if !h.base.acquireConcurrency(ctx, constants.TaskTypePollingPackage) { @@ -52,42 +53,48 @@ func (h *PollingPackageHandler) Handle(ctx context.Context, t *asynq.Task) error return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) } defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingPackage) + return h.base.applyLegacyOutcome(ctx, cardID, constants.TaskTypePollingPackage, h.Step(ctx, cardID)) +} +// Step 执行一次套餐检查步骤（按规则重新判断停复机），返回结果分类；不负责重入队。 +// 调用方负责在 ctx 中设置 Worker 审计上下文。旧分片调度与按运营商账户调度共用。 +func (h *PollingPackageHandler) Step(ctx context.Context, cardID uint) StepOutcome { + startTime := time.Now() // 优先轮询互斥接缝：存在活动优先项时本轮即优先执行或被跳过；无活动项时行为与既有实现一致。 priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingPackage) if priorityErr != nil { h.base.logger.Error("查询活动优先轮询项失败", zap.Uint("card_id", cardID), zap.Error(priorityErr)) h.base.updateStats(ctx, constants.TaskTypePollingPackage, false, time.Since(startTime)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) + return h.base.stepFail(ctx, "", "查询活动优先轮询项失败", priorityErr) } if skip { h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后", zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingPackage)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) + return stepDefer(0) } card, cacheErr := h.base.getCardWithCache(ctx, cardID) if cacheErr != nil { if isNotFound(cacheErr) { priority.FailFinal(ctx, "卡不存在") - return nil + return StepOutcome{Result: StepCardGone} } h.base.logger.Error("获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(cacheErr)) // 本地读卡失败属可恢复失败：与其余三个轮询 handler 统一按尝试上限收敛，避免加急事实被静默丢弃。 priority.MarkExecution(ctx) priority.FailRetryable(ctx, "获取卡信息失败") h.base.updateStats(ctx, constants.TaskTypePollingPackage, false, time.Since(startTime)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) + return h.base.stepFail(ctx, "", "获取卡信息失败", cacheErr) } // 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。 if !h.base.priorityInPollingScope(ctx, priority, card) { - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) + return stepSkip() } if h.stopResumeSvc == nil { priority.FailFinal(ctx, "套餐评估能力未配置") h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) + return stepSkip() } // 停复机评估就是本任务的上游执行动作：先累加尝试次数再发起。 @@ -98,7 +105,7 @@ func (h *PollingPackageHandler) Handle(ctx context.Context, t *asynq.Task) error zap.Uint("card_id", cardID), zap.Error(loadErr)) priority.FailRetryable(ctx, "获取卡信息失败") h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) + return h.base.stepFail(ctx, "", "获取卡信息失败", loadErr) } h.base.logger.Info("套餐检查：执行停复机评估", zap.Uint("card_id", cardID), @@ -119,10 +126,11 @@ func (h *PollingPackageHandler) Handle(ctx context.Context, t *asynq.Task) error zap.Uint("card_id", cardID), zap.Error(evalErr)) priority.FailRetryable(ctx, "停复机评估失败") h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) + outcome := h.base.stepFail(ctx, constants.GatewayOpStart, "停复机评估失败", evalErr) + return outcome } priority.Complete(ctx) h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage) + return stepOK() } diff --git a/internal/task/polling_realname_handler.go b/internal/task/polling_realname_handler.go index 49fec769..0485689a 100644 --- a/internal/task/polling_realname_handler.go +++ b/internal/task/polling_realname_handler.go @@ -28,81 +28,88 @@ func NewPollingRealnameHandler(base *PollingBase, gw *gateway.Client, observatio return &PollingRealnameHandler{base: base, gateway: gw, observation: observation, integration: integration} } -// Handle 处理实名检查任务。 +// Handle 处理实名检查任务（旧分片调度）。按运营商账户调度时直接结束。 func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) error { - startedAt := time.Now() cardID, ok := parseTaskPayload(task.Payload(), h.base.logger) if !ok { return nil } + if h.base.carrierSchedulerActive(ctx) { + return nil + } if !h.base.acquireConcurrency(ctx, constants.TaskTypePollingRealname) { h.base.logger.Debug("并发已满，重新入队", zap.Uint("card_id", cardID)) return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) } defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingRealname) + return h.base.applyLegacyOutcome(ctx, cardID, constants.TaskTypePollingRealname, h.Step(ctx, cardID)) +} +// Step 执行一次实名检查步骤，返回结果分类；不负责重入队。旧分片调度与按运营商账户调度共用。 +func (h *PollingRealnameHandler) Step(ctx context.Context, cardID uint) StepOutcome { + startedAt := time.Now() // 优先轮询互斥接缝：存在活动优先项时本轮即优先执行或被跳过；无活动项时行为与既有实现一致。 priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingRealname) if priorityErr != nil { - return h.failAndRequeue(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr) + return h.fail(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr) } if skip { h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后", zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingRealname)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) + return stepDefer(0) } card, err := h.base.getCardWithCache(ctx, cardID) if err != nil { if isNotFound(err) { priority.FailFinal(ctx, "卡不存在") - return nil + return StepOutcome{Result: StepCardGone} } // 本地读卡失败属可恢复失败：先按真实执行累加尝试次数，再按上限收敛，避免加急事实被静默丢弃。 priority.MarkExecution(ctx) priority.FailRetryable(ctx, "获取卡信息失败") - return h.failAndRequeue(ctx, cardID, startedAt, "获取卡信息失败", err) + return h.fail(ctx, cardID, startedAt, "获取卡信息失败", err) } // 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。 if !h.base.priorityInPollingScope(ctx, priority, card) { - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) + return stepSkip() } if h.gateway == nil { priority.FailFinal(ctx, "实名查询能力未配置") - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) + return stepSkip() } // 所属运营商账户未开启实名管控时实名结果不影响停复机，跳过查询并按正常间隔重入队（不再按行业卡跳过）。 skippable, scopeErr := h.base.realnameQuerySkippable(ctx, card) if scopeErr != nil { priority.FailRetryable(ctx, "查询运营商账户实名管控失败") - return h.failAndRequeue(ctx, cardID, startedAt, "查询运营商账户实名管控失败", scopeErr) + return h.fail(ctx, cardID, startedAt, "查询运营商账户实名管控失败", scopeErr) } if skippable { priority.FailFinal(ctx, "所属运营商账户未开启实名管控，跳过实名查询") h.base.logger.Debug("所属运营商账户未开启实名管控，跳过实名查询", zap.Uint("card_id", cardID), zap.Uint("carrier_id", card.CarrierID)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) + return stepSkip() } priority.MarkExecution(ctx) attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayRealname, constants.CardObservationSceneRealnamePolling) result, err := h.gateway.QueryRealnameStatus(gateway.WithCarrier(ctx, card.CarrierID), &gateway.CardStatusReq{CardNo: card.ICCID}) if gateway.IsRateLimited(err) { priority.FailRetryable(ctx, "运营商账户网关限流，稍后重试") - return h.base.requeueRateLimited(cardID, constants.TaskTypePollingRealname) + return StepOutcome{Result: StepRateLimited} } if err != nil { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询实名状态失败", priority) priority.FailRetryable(ctx, "查询实名状态失败") - return h.failAndRequeue(ctx, cardID, startedAt, "查询实名状态失败", err) + return h.failOp(ctx, cardID, startedAt, constants.GatewayOpQueryRealname, "查询实名状态失败", err) } if strings.TrimSpace(result.ICCID) == "" { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "实名查询响应缺少 ICCID", priority) priority.FailRetryable(ctx, "实名查询响应缺少 ICCID") - return h.failAndRequeue(ctx, cardID, startedAt, "实名查询响应缺少 ICCID", nil) + return h.fail(ctx, cardID, startedAt, "实名查询响应缺少 ICCID", nil) } ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingRealname, "实名状态轮询任务", attempt.IntegrationID) if h.observation == nil { priority.FailFinal(ctx, "卡实名观测能力未配置") - return h.failAndRequeue(ctx, cardID, startedAt, "卡实名观测能力未配置", nil) + return h.fail(ctx, cardID, startedAt, "卡实名观测能力未配置", nil) } now := time.Now() decision, err := h.observation.ApplyCardObservation(ctx, carddomain.RealnameObservation{ @@ -115,7 +122,7 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e if err != nil { _ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用实名观测失败", priority) priority.FailRetryable(ctx, "应用实名观测失败") - return h.failAndRequeue(ctx, cardID, startedAt, "应用实名观测失败", err) + return h.fail(ctx, cardID, startedAt, "应用实名观测失败", err) } changed := decision.StatusChanged if changed { @@ -133,15 +140,19 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e h.base.logger.Info("实名轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric)) priority.Complete(ctx) h.base.updateStats(ctx, constants.TaskTypePollingRealname, true, time.Since(startedAt)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) + return stepOK() } -func (h *PollingRealnameHandler) failAndRequeue(ctx context.Context, cardID uint, startedAt time.Time, message string, err error) error { +func (h *PollingRealnameHandler) fail(ctx context.Context, cardID uint, startedAt time.Time, message string, err error) StepOutcome { + return h.failOp(ctx, cardID, startedAt, constants.GatewayOpQueryRealname, message, err) +} + +func (h *PollingRealnameHandler) failOp(ctx context.Context, cardID uint, startedAt time.Time, op, message string, err error) StepOutcome { fields := []zap.Field{zap.Uint("card_id", cardID)} if err != nil { fields = append(fields, zap.Error(err)) } h.base.logger.Warn(message, fields...) h.base.updateStats(ctx, constants.TaskTypePollingRealname, false, time.Since(startedAt)) - return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname) + return h.base.stepFail(ctx, op, message, err) } diff --git a/internal/task/polling_step.go b/internal/task/polling_step.go new file mode 100644 index 00000000..a77cc1c4 --- /dev/null +++ b/internal/task/polling_step.go @@ -0,0 +1,123 @@ +package task + +import ( + "context" + "time" + + "github.com/break/junhong_cmp_fiber/internal/gateway" + "github.com/break/junhong_cmp_fiber/pkg/constants" +) + +// StepResult 是一个检查步骤的执行结果分类。 +type StepResult int + +const ( + // StepSucceeded 检查成功（含观测入账、评估完成）。 + StepSucceeded StepResult = iota + // StepFailed 检查失败，计入连续失败（是否计入还要看失败类别，见 countsAsFailure）。 + StepFailed + // StepRateLimited 本账户限流拒绝或等待超时，请求未发出：短延迟重排，不计失败。 + StepRateLimited + // StepDeferred 暂不可执行（流量同步锁被占、优先项由其他执行持有）：短延迟或按间隔重排，不计失败。 + StepDeferred + // StepSkipped 本轮无需执行（能力未配置、不在轮询范围、所属账户未开启实名管控）：按正常间隔重排。 + StepSkipped + // StepCardGone 卡不存在：不再重排。 + StepCardGone + // StepStopPolling 观测后该卡已关闭轮询（风险停机、已销户）：不再重排。 + StepStopPolling +) + +// StepOutcome 是一个检查步骤的结果。旧分片调度把它映射为原有的重入队动作；按运营商账户调度据此写回到期与失败计数。 +type StepOutcome struct { + Result StepResult + // DeferDelay 为 StepDeferred 时的建议短延迟；0 表示按正常间隔。 + DeferDelay time.Duration + // Category 失败类别：网关错误类别（network / timeout / http / invalid_response），上游业务错误时为错误分类类别。 + Category string + // Message 失败信息（已脱敏截断）。 + Message string + // Observed 卡状态步骤观测到的网络状态（online / offline），用于延迟确认是否已满足预期。 + Observed string + // TriggeredEvaluation 流量步骤本轮观测产生了流量变化（会触发停复机评估），可据此跳过套餐步骤。 + TriggeredEvaluation bool +} + +// 卡状态步骤的观测结果取值。 +const ( + ObservedOnline = "online" + ObservedOffline = "offline" +) + +func stepOK() StepOutcome { return StepOutcome{Result: StepSucceeded} } + +func stepSkip() StepOutcome { return StepOutcome{Result: StepSkipped} } + +func stepDefer(delay time.Duration) StepOutcome { + return StepOutcome{Result: StepDeferred, DeferDelay: delay} +} + +// stepFail 按错误构造失败结果：网关结构化错误取其类别，上游业务错误再按错误分类规则细分（用于判断是否计入连续失败与告警）。 +func (b *PollingBase) stepFail(ctx context.Context, op, message string, err error) StepOutcome { + outcome := StepOutcome{Result: StepFailed, Message: message, Category: "local"} + if gwErr, ok := gateway.AsGatewayError(err); ok { + outcome.Category = gwErr.Category + if gwErr.Category == constants.GatewayErrorUpstream { + outcome.Message = message + "：" + gwErr.ProviderMessage + if b.errorClassifier != nil { + outcome.Category = b.errorClassifier(ctx, op, gwErr.ProviderMessage) + } + } + } + return outcome +} + +// CountsAsFailure 判断失败结果是否计入连续失败：上游频率限制与已是目标状态不计入。 +func (o StepOutcome) CountsAsFailure() bool { + if o.Result != StepFailed { + return false + } + switch o.Category { + case constants.GatewayClassRateLimitedUpstream, constants.GatewayClassAlreadyTarget: + return false + } + return true +} + +// ErrorClassifier 对上游业务错误信息分类，返回错误分类类别。 +type ErrorClassifier func(ctx context.Context, op, message string) string + +// SetErrorClassifier 注入上游错误分类器（按运营商账户调度的失败退避与告警使用）。 +func (b *PollingBase) SetErrorClassifier(classifier ErrorClassifier) { + b.errorClassifier = classifier +} + +// SchedulerModeReader 返回当前卡轮询调度模式（受控系统配置 polling.scheduler.mode）。 +type SchedulerModeReader func(ctx context.Context) string + +// SetSchedulerModeReader 注入调度模式读取；未注入时视为旧分片调度。 +func (b *PollingBase) SetSchedulerModeReader(reader SchedulerModeReader) { + b.schedulerMode = reader +} + +// carrierSchedulerActive 判断当前是否已切到按运营商账户调度：是则旧 Asynq 轮询任务直接结束，不调网关、不重入旧队列。 +func (b *PollingBase) carrierSchedulerActive(ctx context.Context) bool { + return b != nil && b.schedulerMode != nil && b.schedulerMode(ctx) == constants.PollingSchedulerModeCarrier +} + +// applyLegacyOutcome 把步骤结果映射为旧分片调度原有的重入队动作。 +func (b *PollingBase) applyLegacyOutcome(ctx context.Context, cardID uint, taskType string, outcome StepOutcome) error { + switch outcome.Result { + case StepCardGone, StepStopPolling: + return nil + case StepRateLimited: + return b.requeueRateLimited(cardID, taskType) + case StepDeferred: + if outcome.DeferDelay > 0 { + return b.queueMgr.Requeue(ctx, cardID, taskType, time.Now().Add(outcome.DeferDelay)) + } + return b.requeueCard(ctx, cardID, taskType) + default: + return b.requeueCard(ctx, cardID, taskType) + } +} diff --git a/pkg/queue/handler.go b/pkg/queue/handler.go index dc7409f6..22b07361 100644 --- a/pkg/queue/handler.go +++ b/pkg/queue/handler.go @@ -40,6 +40,7 @@ type Handler struct { workerResult *WorkerBootstrapResult asynqClient *asynq.Client pollingBase *task.PollingBase + pollingSteps PollingStepHandlers pollingStopResumeSvc iot_card_svc.StopResumeServiceInterface } @@ -292,6 +293,7 @@ func (h *Handler) registerPollingHandlers() { cardStatusHandler := task.NewPollingCardStatusHandler( h.pollingBase, h.gatewayClient, h.workerResult.Services.CardObservation, integrationRepository) + h.pollingSteps = PollingStepHandlers{Realname: realnameHandler, Carddata: carddataHandler, Package: packageHandler, CardStatus: cardStatusHandler} h.mux.HandleFunc(constants.TaskTypePollingRealname, realnameHandler.Handle) h.mux.HandleFunc(constants.TaskTypePollingCarddata, carddataHandler.Handle) h.mux.HandleFunc(constants.TaskTypePollingPackage, packageHandler.Handle) @@ -408,3 +410,16 @@ func (h *Handler) registerOperationsReportSnapshotHandler() { func (h *Handler) GetMux() *asynq.ServeMux { return h.mux } + +// PollingStepHandlers 是四个轮询检查步骤处理器，旧分片调度注册为 Asynq handler，按运营商账户调度的卡任务执行器复用其 Step。 +type PollingStepHandlers struct { + Realname *task.PollingRealnameHandler + Carddata *task.PollingCarddataHandler + Package *task.PollingPackageHandler + CardStatus *task.PollingCardStatusHandler +} + +// PollingSteps 返回轮询检查步骤处理器（RegisterHandlers 之后可用）。 +func (h *Handler) PollingSteps() PollingStepHandlers { + return h.pollingSteps +}