diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 9185b212..82ea8fb7 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -1045,6 +1045,8 @@ func registerAsynqScheduleTasks(asynqScheduler *asynq.Scheduler, auditArchiveEna nil, asynq.MaxRetry(3), asynq.Timeout(2*time.Minute), + // 同一时刻只保留一个恢复任务:队列拥堵时不再堆积重复的恢复任务。 + asynq.Unique(2*time.Minute), asynq.Queue(constants.QueueForTaskType(constants.TaskTypeWeComApprovalRecovery)), )); err != nil { return fmt.Errorf("注册企业微信审批主动恢复定时任务失败: %w", err) diff --git a/internal/infrastructure/wecom/approval_context_repository.go b/internal/infrastructure/wecom/approval_context_repository.go index ccfbbd00..47e7ca6b 100644 --- a/internal/infrastructure/wecom/approval_context_repository.go +++ b/internal/infrastructure/wecom/approval_context_repository.go @@ -159,6 +159,8 @@ func (r *ApprovalContextRepository) ListUnknownRecovery(ctx context.Context, cut err := r.db.WithContext(ctx).Table("tb_wecom_approval_context AS wc"). Select(`wc.approval_instance_id AS instance_id, wc.application_id, wc.template_id, wc.creator_userid, wc.sp_no, wc.submission_status, COALESCE(wc.submission_attempted_at, wc.updated_at) AS submission_attempted_at`). + // 只恢复启用中企业微信应用的审批:已停用或已删除的应用调用必然失败,反复投递会堆积任务。 + Joins("JOIN tb_wecom_application AS wa ON wa.id = wc.application_id AND wa.status = ? AND wa.deleted_at IS NULL", constants.StatusEnabled). Where("wc.submission_status = ? AND wc.sp_no = '' AND (wc.last_recovery_at IS NULL OR wc.last_recovery_at <= ?)", constants.WeComSubmissionStatusUnknown, cutoff.UTC()). Order("wc.id ASC").Limit(limit).Scan(&records).Error @@ -191,6 +193,8 @@ func (r *ApprovalContextRepository) ListPendingSync(ctx context.Context, cutoff Select(`wc.approval_instance_id AS instance_id, wc.application_id, wc.template_id, wc.creator_userid, wc.sp_no, wc.submission_status, COALESCE(wc.submission_attempted_at, wc.updated_at) AS submission_attempted_at`). Joins("JOIN tb_approval_instance AS ai ON ai.id = wc.approval_instance_id"). + // 只轮询启用中企业微信应用的审批:已停用或已删除的应用调用必然失败,反复投递会堆积任务。 + Joins("JOIN tb_wecom_application AS wa ON wa.id = wc.application_id AND wa.status = ? AND wa.deleted_at IS NULL", constants.StatusEnabled). Where("wc.submission_status = ? AND wc.sp_no <> '' AND ai.status = ? AND (wc.last_synced_at IS NULL OR wc.last_synced_at <= ?)", constants.WeComSubmissionStatusSubmitted, constants.ApprovalStatusPending, cutoff.UTC()). Order("COALESCE(wc.last_synced_at, wc.created_at) ASC, wc.id ASC").Limit(limit).Scan(&records).Error diff --git a/internal/infrastructure/wecom/approval_detail_task.go b/internal/infrastructure/wecom/approval_detail_task.go index aadeeb55..2fc5a31f 100644 --- a/internal/infrastructure/wecom/approval_detail_task.go +++ b/internal/infrastructure/wecom/approval_detail_task.go @@ -2,6 +2,8 @@ package wecom import ( "context" + stderrors "errors" + "fmt" "strconv" "strings" "time" @@ -53,6 +55,11 @@ func (h *ApprovalDetailTaskHandler) Handle(ctx context.Context, task *asynq.Task } detail, err := h.details.Get(ctx, payload.ApplicationID, payload.SPNo, resourceID, correlationID) if err != nil { + // 企业微信应用已停用或不存在:重试必然失败,直接结束不重试(启用后由轮询重新投递)。 + var appErr *errors.AppError + if stderrors.As(err, &appErr) && appErr.Code == errors.CodeWeComApplicationNotFound { + return fmt.Errorf("%w: %v", asynq.SkipRetry, err) + } return err } if record == nil { diff --git a/internal/infrastructure/wecom/approval_recovery_task.go b/internal/infrastructure/wecom/approval_recovery_task.go index d1f1ff0c..3efaece8 100644 --- a/internal/infrastructure/wecom/approval_recovery_task.go +++ b/internal/infrastructure/wecom/approval_recovery_task.go @@ -2,6 +2,7 @@ package wecom import ( "context" + stderrors "errors" "strings" "time" @@ -176,8 +177,14 @@ func (h *ApprovalRecoveryTaskHandler) findUniqueSPNo(ctx context.Context, record return unbound[0], integrationIDs, nil } +// enqueueDetailSync 投递轮询详情同步:同一应用同一审批单在唯一性锁有效期内只保留一个任务(锁在任务成功或到期后释放, +// 归档的失败任务不会永久占用),轮询每轮都会再次投递,失败只少量重试。 func (h *ApprovalRecoveryTaskHandler) enqueueDetailSync(ctx context.Context, applicationID uint, spNo string) error { - return h.queue.EnqueueTask(ctx, constants.TaskTypeWeComApprovalSync, ApprovalDetailSyncTask{ + err := h.queue.EnqueueTask(ctx, constants.TaskTypeWeComApprovalSync, ApprovalDetailSyncTask{ ApplicationID: applicationID, SPNo: spNo, Source: constants.ApprovalSyncSourcePolling, - }) + }, asynq.Unique(constants.WeComApprovalPollingSyncUnique), asynq.MaxRetry(constants.WeComApprovalPollingSyncMaxRetry)) + if stderrors.Is(err, asynq.ErrTaskIDConflict) || stderrors.Is(err, asynq.ErrDuplicateTask) { + return nil + } + return err } diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index e897f696..72944077 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -84,15 +84,23 @@ const ( TaskTypeAutoPurchaseAfterRecharge = "task:auto_purchase_after_recharge" // 充值后自动购包 // 定时任务类型(由 Asynq Scheduler 调度) - TaskTypeAlertCheck = "alert:check" // 告警检查 - TaskTypeDataCleanup = "data:cleanup" // 数据清理 - TaskTypeNotificationCleanup = "notification:cleanup" // 站内通知保留清理 - TaskTypePackageExpiryReminder = "package:expiry:reminder" // 每日套餐临期提醒扫描 - TaskTypeDailyTrafficFlush = "traffic:daily:flush" // 每日流量落盘 - TaskTypeOutboxDeliver = "outbox:deliver" // 公共 Outbox 事件投递 - TaskTypeCardObservationSeries = "card_observation:series" // 卡观测事件序列尝试 - TaskTypeWeComApprovalSync = "wecom:approval:sync" // 企业微信审批详情异步同步 - TaskTypeWeComApprovalRecovery = "wecom:approval:recovery" // 企业微信审批主动恢复与轮询 + TaskTypeAlertCheck = "alert:check" // 告警检查 + TaskTypeDataCleanup = "data:cleanup" // 数据清理 + TaskTypeNotificationCleanup = "notification:cleanup" // 站内通知保留清理 + TaskTypePackageExpiryReminder = "package:expiry:reminder" // 每日套餐临期提醒扫描 + TaskTypeDailyTrafficFlush = "traffic:daily:flush" // 每日流量落盘 + TaskTypeOutboxDeliver = "outbox:deliver" // 公共 Outbox 事件投递 + TaskTypeCardObservationSeries = "card_observation:series" // 卡观测事件序列尝试 + TaskTypeWeComApprovalSync = "wecom:approval:sync" // 企业微信审批详情异步同步 + TaskTypeWeComApprovalRecovery = "wecom:approval:recovery" // 企业微信审批主动恢复与轮询 + // WeComApprovalPollingSyncMaxRetry 轮询投递的审批详情同步最大重试次数:轮询每轮都会再次投递,无需长重试。 + WeComApprovalPollingSyncMaxRetry = 3 +) + +// WeComApprovalPollingSyncUnique 轮询投递的审批详情同步唯一性锁有效期:期间同一审批单不重复投递。 +const WeComApprovalPollingSyncUnique = 10 * time.Minute + +const ( TaskTypeAgentRechargeRecovery = "agent_recharge:payment:recovery" // 代理在线充值支付恢复与查单 TaskTypeRefundChannelRecovery = "refund:channel:recovery" // 渠道原路退款结果恢复与查询 TaskTypeRefundCommissionRecovery = "refund:commission:recovery" // 退款佣金回溯后处理补偿