1214181e78
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 16m2s
- 轮询步骤异步模式下只调网关并发布 card.ingest 入库消息，入库消费者固定并发 30、按采集 ID 去重、可重试错误不限次数退避重投 - 采集记录经 card.collection 攒批写入，失败整批重投，新增 (observation_id, observed_at) 唯一索引 - 网络、实名观测增加乱序保护；轮询读查询走缓存，套餐状态查询各自计时且失败不缓存 - 入库积压告警与反压；系统配置 polling.ingest.mode 默认 sync 可回退；测试环境 nsqd 消息全部落盘 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
306 lines
12 KiB
Go
306 lines
12 KiB
Go
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 !e.base.scopeAllowsCached(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 || e.base.flowEvaluatedRecently(ctx, msg.CardID)) {
|
||
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:
|
||
// 跳过（如所属账户未开实名管控）视同本轮已检查：记最近成功时间，使请求检查的冷却同样生效。
|
||
if outcome.Result == StepSkipped {
|
||
set[carrierqueue.FieldOK(checkType)] = strconv.FormatInt(now.UnixMilli(), 10)
|
||
}
|
||
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])
|
||
}
|