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>
384 lines
16 KiB
Go
384 lines
16 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"strconv"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/pkg/auditfailure"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// pollingPriorityTaskNames 是纳入优先轮询的既有任务类型的中文名，用于日志与审计展示。
|
||
var pollingPriorityTaskNames = map[string]string{
|
||
constants.TaskTypePollingRealname: "实名状态轮询任务",
|
||
constants.TaskTypePollingCarddata: "卡流量轮询任务",
|
||
constants.TaskTypePollingPackage: "套餐状态轮询任务",
|
||
constants.TaskTypePollingCardStatus: "卡网络状态轮询任务",
|
||
}
|
||
|
||
// pollingPriorityRound 承载本轮轮询执行是否同时是一次优先执行。
|
||
//
|
||
// 无活动优先项时接缝返回 nil，其所有方法都是无操作，因此调用方在「无优先项」路径上
|
||
// 完全保持既有行为（入队、出队、移除与延后一行不改）。领取成功时必须恰好调用一次结果收口方法
|
||
// （Complete / FailRetryable / FailFinal），以保证不出现「领了优先项却没执行也没落终态或退回活动态」的路径。
|
||
//
|
||
// 上下文约定（照 PollingBase 既有约定）：本类型的写库与审计一律忽略调用方 ctx，
|
||
// 内部改用 pollingFallbackContext 的独立短超时上下文。Asynq 任务 ctx 超时或取消后，
|
||
// 失败事实、终态与审计仍必须尽量落下，否则优先项会停在执行中只能等租约接管。
|
||
type pollingPriorityRound struct {
|
||
base *PollingBase
|
||
item *model.PollingPriorityItem
|
||
taskType string
|
||
taskName string
|
||
// attempts 与库内 attempt_count 保持同步，用于判定是否达到尝试上限。
|
||
attempts int
|
||
}
|
||
|
||
// beginPollingPriority 轮询共享基类的优先项认领接缝。
|
||
//
|
||
// 位置约定：必须在取并发信号量之后、调用上游之前（carddata 还需在流量同步锁获取成功之后）。
|
||
// 返回语义：
|
||
// - round 非 nil：本轮领取成功，本轮执行同时就是优先执行；
|
||
// - skip 为 true：存在活动优先项但本轮必须跳过并延后（已被其它执行持有，或超租约接管未命中），
|
||
// 调用方必须释放并发、按既有间隔 requeueCard 延后、零上游调用、不写失败审计；
|
||
// - err 非 nil：活动项探测失败，调用方按既有失败分支处理；
|
||
// - round 为 nil 且 skip 为 false：无活动优先项，调用方后续行为与既有实现完全一致。
|
||
func (b *PollingBase) beginPollingPriority(ctx context.Context, cardID uint, taskType string) (*pollingPriorityRound, bool, error) {
|
||
if b == nil || b.priorityStore == nil {
|
||
return nil, false, nil
|
||
}
|
||
if b.priorityIndexAbsent(ctx, cardID, taskType) {
|
||
return nil, false, nil
|
||
}
|
||
item, err := b.priorityStore.FindActive(ctx, cardID, taskType)
|
||
if err != nil {
|
||
return nil, false, err
|
||
}
|
||
if item == nil {
|
||
return nil, false, nil
|
||
}
|
||
|
||
now := time.Now()
|
||
switch item.Status {
|
||
case constants.PollingPriorityStatusPending:
|
||
// 待执行项必须由到达的执行认领并执行，否则提示通道产生的执行会被自己跳过、优先项永远无人执行。
|
||
claimed, claimErr := b.priorityStore.ClaimPending(ctx, cardID, taskType, now)
|
||
if claimErr != nil {
|
||
return nil, false, claimErr
|
||
}
|
||
if claimed == nil {
|
||
return nil, true, nil
|
||
}
|
||
return b.newPriorityRound(ctx, claimed, taskType, now), false, nil
|
||
case constants.PollingPriorityStatusProcessing:
|
||
// 执行中且认领租约未到期：不发第二次上游调用。
|
||
if item.ClaimedAt != nil && now.Before(item.ClaimedAt.Add(time.Duration(constants.PollingPriorityClaimLease)*time.Second)) {
|
||
return nil, true, nil
|
||
}
|
||
// 超过租约或认领时间缺失视为崩溃遗留：允许相邻执行接管，该卡该任务类型不被永久跳过。
|
||
takenOver, takeoverErr := b.priorityStore.TakeOverExpiredClaim(ctx, cardID, taskType, now, constants.PollingPriorityClaimLease)
|
||
if takeoverErr != nil {
|
||
return nil, false, takeoverErr
|
||
}
|
||
if takenOver == nil {
|
||
return nil, true, nil
|
||
}
|
||
return b.newPriorityRound(ctx, takenOver, taskType, now), false, nil
|
||
default:
|
||
// 终态行不属于活动集合，FindActive 不会返回；防御性按跳过处理。
|
||
return nil, true, nil
|
||
}
|
||
}
|
||
|
||
// newPriorityRound 创建领取成功后的执行轮次，并记录领取事实。
|
||
func (b *PollingBase) newPriorityRound(ctx context.Context, item *model.PollingPriorityItem, taskType string, now time.Time) *pollingPriorityRound {
|
||
round := &pollingPriorityRound{
|
||
base: b,
|
||
item: item,
|
||
taskType: taskType,
|
||
taskName: pollingPriorityTaskName(taskType),
|
||
attempts: item.AttemptCount,
|
||
}
|
||
before := round.stateData()
|
||
item.Status = constants.PollingPriorityStatusProcessing
|
||
item.ClaimedAt = &now
|
||
round.writeAudit(constants.AuditActionPollingPriorityClaimed, "卡轮询优先项已领取", constants.AuditResultSuccess, "", "", before)
|
||
return round
|
||
}
|
||
|
||
// priorityInPollingScope 执行前校验该卡是否仍在轮询范围内：卡自身与绑定设备的轮询开关均为启用。
|
||
//
|
||
// 优先通道不得重启已停止轮询的卡；既有人工触发入口的范围语义不受本方法影响。
|
||
// 无活动优先项（round 为 nil）时恒为真，普通轮询保持既有行为；不在范围内时按业务校验类失败
|
||
// 终结优先项并返回 false，调用方据此跳过上游调用。
|
||
func (b *PollingBase) priorityInPollingScope(_ context.Context, round *pollingPriorityRound, card *model.IotCard) bool {
|
||
if round == nil {
|
||
return true
|
||
}
|
||
if card == nil {
|
||
round.FailFinal(nil, "卡不存在")
|
||
return false
|
||
}
|
||
scopeCtx, cancel := pollingFallbackContext()
|
||
inScope := b.scopeChecker == nil || b.scopeChecker.ShouldEnqueue(scopeCtx, card)
|
||
cancel()
|
||
if inScope {
|
||
return true
|
||
}
|
||
round.FailFinal(nil, "卡已不在轮询范围内")
|
||
return false
|
||
}
|
||
|
||
// MarkExecution 标记即将真正发起上游调用，并在调用之前累加尝试次数与执行开始时间。
|
||
// 崩溃遗留的执行中项被接管后据此按尝试上限收敛。
|
||
func (r *pollingPriorityRound) MarkExecution(_ context.Context) {
|
||
if r == nil {
|
||
return
|
||
}
|
||
ctx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
now := time.Now()
|
||
hit, err := r.base.priorityStore.MarkExecutionStarted(ctx, r.item.ID, now)
|
||
if err != nil {
|
||
r.base.logger.Warn("累加优先轮询项尝试次数失败",
|
||
zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
|
||
}
|
||
if !hit {
|
||
r.base.logger.Warn("优先轮询项已不处于执行中，尝试次数未累加",
|
||
zap.Uint("priority_item_id", r.item.ID))
|
||
}
|
||
r.attempts = r.item.AttemptCount + 1
|
||
r.item.AttemptCount = r.attempts
|
||
r.item.StartedAt = &now
|
||
r.item.FinishedAt = nil
|
||
r.item.NextRunAt = nil
|
||
}
|
||
|
||
// RecordIntegrationLog 把本次执行对应的集成交互日志标识写回优先项。
|
||
// 只保存日志标识（不复制凭证或响应原文）；写入失败只损失溯源细节，不阻断执行。
|
||
func (r *pollingPriorityRound) RecordIntegrationLog(integrationLogID string) {
|
||
if r == nil || r.item == nil || integrationLogID == "" {
|
||
return
|
||
}
|
||
ctx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
if err := r.base.priorityStore.RecordIntegrationLog(ctx, r.item.ID, integrationLogID); err != nil {
|
||
r.base.logger.Warn("记录优先轮询项集成交互日志标识失败",
|
||
zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
|
||
return
|
||
}
|
||
r.item.IntegrationLogID = integrationLogID
|
||
}
|
||
|
||
// nextRunAt 计算该项退回活动态后的下一次计划执行时间，复用与重入队完全相同的间隔解析。
|
||
func (r *pollingPriorityRound) nextRunAt() time.Time {
|
||
if r == nil || r.base == nil {
|
||
return time.Now().Add(pollingRequeueFallbackInterval)
|
||
}
|
||
ctx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
nextCheckAt, ok := r.base.pollingNextCheckAt(ctx, r.item.CardID, r.taskType)
|
||
if !ok {
|
||
// 该任务类型轮询间隔为 NULL：普通轮询不会接管，仍写下兜底时间让重试路径可见。
|
||
return time.Now().Add(pollingRequeueFallbackInterval)
|
||
}
|
||
return nextCheckAt
|
||
}
|
||
|
||
// Complete 记录优先项执行成功并出队，该卡继续按普通轮询运行。
|
||
func (r *pollingPriorityRound) Complete(_ context.Context) {
|
||
if r == nil {
|
||
return
|
||
}
|
||
ctx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
before := r.stateData()
|
||
finishedAt := time.Now()
|
||
hit, err := r.base.priorityStore.MarkCompleted(ctx, r.item.ID)
|
||
if err != nil {
|
||
r.base.logger.Warn("更新优先轮询项完成状态失败",
|
||
zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
|
||
}
|
||
if !hit {
|
||
r.base.logger.Warn("优先轮询项已完成或不处于执行中，跳过完成收口",
|
||
zap.Uint("priority_item_id", r.item.ID))
|
||
return
|
||
}
|
||
r.item.Status = constants.PollingPriorityStatusCompleted
|
||
r.item.Result = constants.PollingPriorityResultSuccess
|
||
r.item.FailureReason = ""
|
||
r.item.FinishedAt = &finishedAt
|
||
r.item.DequeuedAt = &finishedAt
|
||
r.item.NextRunAt = nil
|
||
r.writeAudit(constants.AuditActionPollingPriorityCompleted, "卡轮询优先项执行完成并出队", constants.AuditResultSuccess, "", "", before)
|
||
}
|
||
|
||
// FailRetryable 记录一次可恢复失败（上游超时、失败或响应无效）。
|
||
// 未达尝试上限时退回活动态并写下一次计划执行时间；达到上限时记录失败终态并出队。
|
||
func (r *pollingPriorityRound) FailRetryable(_ context.Context, reason string) {
|
||
if r == nil {
|
||
return
|
||
}
|
||
ctx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
before := r.stateData()
|
||
finishedAt := time.Now()
|
||
if r.attempts < constants.PollingPriorityMaxAttempts {
|
||
nextRunAt := r.nextRunAt()
|
||
hit, err := r.base.priorityStore.ReturnToPending(ctx, r.item.ID, reason, nextRunAt)
|
||
if err != nil {
|
||
r.base.logger.Warn("退回优先轮询项活动状态失败",
|
||
zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
|
||
}
|
||
if !hit {
|
||
r.base.logger.Warn("优先轮询项不处于执行中，跳过退回活动态收口",
|
||
zap.Uint("priority_item_id", r.item.ID))
|
||
return
|
||
}
|
||
r.item.Status = constants.PollingPriorityStatusPending
|
||
r.item.Result = constants.PollingPriorityResultFailed
|
||
r.item.FailureReason = reason
|
||
r.item.ClaimedAt = nil
|
||
r.item.FinishedAt = &finishedAt
|
||
r.item.NextRunAt = &nextRunAt
|
||
r.writeAudit(constants.AuditActionPollingPriorityFailed, "卡轮询优先项执行失败", constants.AuditResultFailed, "", reason, before)
|
||
r.writeAudit(constants.AuditActionPollingPriorityRetried, "卡轮询优先项未达尝试上限，退回活动态重新排期", constants.AuditResultFailed, "", reason, before)
|
||
return
|
||
}
|
||
|
||
hit, err := r.base.priorityStore.MarkFailed(ctx, r.item.ID, reason)
|
||
if err != nil {
|
||
r.base.logger.Warn("更新优先轮询项失败状态失败",
|
||
zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
|
||
}
|
||
if !hit {
|
||
r.base.logger.Warn("优先轮询项不处于执行中，跳过失败出队收口",
|
||
zap.Uint("priority_item_id", r.item.ID))
|
||
return
|
||
}
|
||
r.item.Status = constants.PollingPriorityStatusFailed
|
||
r.item.Result = constants.PollingPriorityResultFailed
|
||
r.item.FailureReason = reason
|
||
r.item.FinishedAt = &finishedAt
|
||
r.item.DequeuedAt = &finishedAt
|
||
r.item.NextRunAt = nil
|
||
r.writeAudit(constants.AuditActionPollingPriorityFailed, "卡轮询优先项执行失败", constants.AuditResultFailed, "", reason, before)
|
||
r.writeAudit(constants.AuditActionPollingPriorityDequeued, "卡轮询优先项达到尝试上限，失败出队", constants.AuditResultFailed, "", reason, before)
|
||
}
|
||
|
||
// FailFinal 记录业务校验类失败（卡不存在、不在轮询范围、能力未配置等）：不重试，直接失败出队。
|
||
func (r *pollingPriorityRound) FailFinal(_ context.Context, reason string) {
|
||
if r == nil {
|
||
return
|
||
}
|
||
ctx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
before := r.stateData()
|
||
hit, err := r.base.priorityStore.MarkFailed(ctx, r.item.ID, reason)
|
||
if err != nil {
|
||
r.base.logger.Warn("更新优先轮询项失败状态失败",
|
||
zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
|
||
}
|
||
if !hit {
|
||
r.base.logger.Warn("优先轮询项不处于执行中，跳过失败出队收口",
|
||
zap.Uint("priority_item_id", r.item.ID))
|
||
return
|
||
}
|
||
finishedAt := time.Now()
|
||
r.item.Status = constants.PollingPriorityStatusFailed
|
||
r.item.Result = constants.PollingPriorityResultFailed
|
||
r.item.FailureReason = reason
|
||
r.item.FinishedAt = &finishedAt
|
||
r.item.DequeuedAt = &finishedAt
|
||
r.item.NextRunAt = nil
|
||
r.writeAudit(constants.AuditActionPollingPriorityFailed, "卡轮询优先项执行业务校验失败", constants.AuditResultFailed, "", reason, before)
|
||
r.writeAudit(constants.AuditActionPollingPriorityDequeued, "卡轮询优先项失败出队且不重试", constants.AuditResultFailed, "", reason, before)
|
||
}
|
||
|
||
// stateData 返回当前优先项状态的前后值快照，覆盖触发来源、操作来源、时间与执行结果。
|
||
func (r *pollingPriorityRound) stateData() map[string]any {
|
||
return map[string]any{
|
||
"status": r.item.Status,
|
||
"result": r.item.Result,
|
||
"trigger_type": r.item.TriggerType,
|
||
"trigger_types": r.item.TriggerTypes,
|
||
"trigger_count": r.item.TriggerCount,
|
||
"last_triggered_at": r.item.LastTriggeredAt,
|
||
"attempt_count": r.item.AttemptCount,
|
||
"attempt_limit": r.item.AttemptLimit,
|
||
"claimed_at": r.item.ClaimedAt,
|
||
"started_at": r.item.StartedAt,
|
||
"finished_at": r.item.FinishedAt,
|
||
"next_run_at": r.item.NextRunAt,
|
||
"dequeued_at": r.item.DequeuedAt,
|
||
"integration_log_id": r.item.IntegrationLogID,
|
||
"failure_reason": r.item.FailureReason,
|
||
"source": constants.AuditSourceWorker,
|
||
"occurred_at": time.Now(),
|
||
}
|
||
}
|
||
|
||
// writeAudit 以独立短事务写入一条优先项生命周期审计。
|
||
// 审计写入失败不阻断轮询执行，按既有约定登记为失败审计二次写入故障。
|
||
func (r *pollingPriorityRound) writeAudit(
|
||
actionCode, summary, result, errorCode, errorSummary string,
|
||
before map[string]any,
|
||
) {
|
||
base := r.base
|
||
if base == nil || base.priorityAudit == nil || base.db == nil || r.item == nil || r.item.ID == 0 {
|
||
return
|
||
}
|
||
ctx, cancel := pollingFallbackContext()
|
||
defer cancel()
|
||
input := audit.TaskInput{
|
||
ActionCode: actionCode, Summary: summary,
|
||
TaskID: r.item.ID, TaskNo: pollingPriorityResourceKey(r.item), DisplayName: "卡轮询优先项",
|
||
Actor: audit.ActorInput{
|
||
Kind: constants.AuditActorSystemTask, ID: r.taskType, Name: r.taskName,
|
||
},
|
||
Source: constants.AuditSourceWorker,
|
||
ScopeType: constants.AuditScopePlatform,
|
||
Result: result,
|
||
ErrorCode: errorCode,
|
||
ErrorSummary: errorSummary,
|
||
CorrelationID: r.taskType,
|
||
IdentitySnapshot: map[string]any{
|
||
"id": r.item.ID, "card_id": r.item.CardID, "task_type": r.item.TaskType,
|
||
"status": r.item.Status, "trigger_type": r.item.TriggerType,
|
||
"trigger_count": r.item.TriggerCount, "manual_operator_id": r.item.ManualOperatorID,
|
||
},
|
||
BeforeData: before,
|
||
AfterData: r.stateData(),
|
||
}
|
||
err := base.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
return base.priorityAudit.WriteTask(ctx, tx, input)
|
||
})
|
||
if err != nil {
|
||
auditfailure.RecordSecondaryWriteFailure(actionCode, input.TaskNo, "", r.taskType, errorCode, err)
|
||
base.logger.Error("优先轮询项审计写入失败",
|
||
zap.String("action", actionCode), zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// pollingPriorityTaskName 返回优先轮询任务类型的中文名，未知类型回退为完整任务类型。
|
||
func pollingPriorityTaskName(taskType string) string {
|
||
if name, ok := pollingPriorityTaskNames[taskType]; ok {
|
||
return name
|
||
}
|
||
return taskType
|
||
}
|
||
|
||
// pollingPriorityResourceKey 返回优先项在审计里的稳定资源键。
|
||
func pollingPriorityResourceKey(item *model.PollingPriorityItem) string {
|
||
if item == nil {
|
||
return ""
|
||
}
|
||
return strconv.FormatUint(uint64(item.ID), 10)
|
||
}
|