2fe7c1f8ac
- 四个轮询处理器拆为步骤函数 Step（返回结果分类），旧 Asynq Handle 只做并发控制与原有重入队映射，行为不变；carrier 模式下旧 handler 直接结束
- 网关限流增加等待模式（ctx 标记），NSQ 消费者等名额到单步超时，超时按限流处理不计失败；记录限流等待次数
- 卡任务执行器：核对 token 并抢执行权，按配置步骤顺序串行执行到期检查（单步超时、总超时、panic 恢复、流量已触发评估时跳过套餐），执行中续期并 Touch，续期失败立即停止；收尾写回到期、连续失败、退避、延迟确认与告警
- 分发器：每秒按账户抢分发锁领取（执行上限按消费能力收紧、优先占比、按限流组剩余 QPS 估算），发布到 polling.card.{账户}，发布失败释放；消费旧优先提示与手动触发；心跳与处理中标记到期计数
- NSQ 生产者与按账户动态订阅的消费者管理器，优雅退出宽限期
- 失败告警异步写 tb_polling_failure_alert（同卡同类型一条未处理，成功转已恢复），账户窗口内告警数超阈值产生账户级告警
- 模式监听：切 carrier 时旧分片调度停止出队，加锁初始化新结构后开始领取并订阅；回 legacy 时停止消费、重建旧分片队列；生命周期（API 与 Worker）在 carrier 模式写账户队列
124 lines
4.9 KiB
Go
124 lines
4.9 KiB
Go
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)
|
||
}
|
||
}
|