From 3618fc2b203e67bd14ef182503a18ef7f1fc56b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=82=B9=E8=BF=9B?= Date: Mon, 26 Oct 2026 11:16:33 +0900 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BC=81=E4=B8=9A=E5=BE=AE=E4=BF=A1?= =?UTF-8?q?=E5=AE=A1=E6=89=B9=E5=90=8C=E6=AD=A5=E4=BB=BB=E5=8A=A1=E5=9C=A8?= =?UTF-8?q?=E5=B7=B2=E5=81=9C=E7=94=A8=E5=BA=94=E7=94=A8=E4=B8=8B=E6=97=A0?= =?UTF-8?q?=E9=99=90=E5=A0=86=E7=A7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 恢复任务每 2 分钟为未终态审批投递详情同步，已停用应用的审批同步必然失败并重试 25 次， 且同一审批单每轮重复投递，测试环境 wecom:approval 队列积压约 15 万任务，在用应用的审批结果迟迟不生效。 - 轮询与结果未知恢复只取启用中企业微信应用的审批 - 轮询投递的详情同步按审批单加 10 分钟唯一性锁，最多重试 3 次 - 应用不存在或已停用时同步任务直接结束不重试 - 恢复定时任务加唯一性，队列拥堵时不再堆积重复恢复任务 Co-Authored-By: Claude Opus 5.5 --- cmd/worker/main.go | 2 ++ .../wecom/approval_context_repository.go | 4 +++ .../wecom/approval_detail_task.go | 7 +++++ .../wecom/approval_recovery_task.go | 11 ++++++-- pkg/constants/constants.go | 26 ++++++++++++------- 5 files changed, 39 insertions(+), 11 deletions(-) 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" // 退款佣金回溯后处理补偿