diff --git a/internal/gateway/client.go b/internal/gateway/client.go index b57ed20..20a67cb 100644 --- a/internal/gateway/client.go +++ b/internal/gateway/client.go @@ -17,7 +17,7 @@ import ( ) const ( - defaultTimeout = 30 * time.Second + defaultTimeout = 60 * time.Second maxIdleConns = 100 maxIdleConnsPerHost = 10 idleConnTimeout = 90 * time.Second diff --git a/internal/polling/package_activation_handler.go b/internal/polling/package_activation_handler.go index 48a9a1f..9e058c1 100644 --- a/internal/polling/package_activation_handler.go +++ b/internal/polling/package_activation_handler.go @@ -390,7 +390,7 @@ func (h *PackageActivationHandler) enqueueActivationTask(ctx context.Context, pa task := asynq.NewTask(constants.TaskTypePackageQueueActivation, payloadBytes, asynq.MaxRetry(3), - asynq.Timeout(30*time.Second), + asynq.Timeout(60*time.Second), asynq.Queue(constants.QueueForTaskType(constants.TaskTypePackageQueueActivation)), ) diff --git a/internal/polling/scheduler.go b/internal/polling/scheduler.go index 1ec2436..9d2dbcb 100644 --- a/internal/polling/scheduler.go +++ b/internal/polling/scheduler.go @@ -262,7 +262,7 @@ func (s *Scheduler) enqueueBatch(ctx context.Context, taskType string, cardIDs [ } task := asynq.NewTask(taskType, payloadBytes, asynq.MaxRetry(0), - asynq.Timeout(30*time.Second), + asynq.Timeout(60*time.Second), asynq.Queue(constants.QueueForTaskType(taskType)), ) if _, err := s.queueClient.Enqueue(task); err != nil { diff --git a/internal/task/polling_base.go b/internal/task/polling_base.go index 4da8b4a..0ec5942 100644 --- a/internal/task/polling_base.go +++ b/internal/task/polling_base.go @@ -29,6 +29,14 @@ var acquireConcurrencyScript = redis.NewScript(` `) const cardTrafficSyncLockTTL = 10 * time.Minute +const pollingFallbackOperationTimeout = 5 * time.Second + +// pollingFallbackContext 创建轮询兜底操作使用的独立短超时上下文。 +// Asynq 任务 ctx 超时后，释放并发计数、释放锁、重入队仍必须尽量完成， +// 否则会造成并发计数短期卡住，甚至卡已出队但未重新入队。 +func pollingFallbackContext() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), pollingFallbackOperationTimeout) +} // PollingBase 轮询共享基类 // 封装并发控制、卡缓存、重入队、配置间隔查询等公共方法，所有 Handler 共享 @@ -86,10 +94,13 @@ func (b *PollingBase) acquireConcurrency(ctx context.Context, taskType string) b } // releaseConcurrency 释放并发信号量 -func (b *PollingBase) releaseConcurrency(ctx context.Context, taskType string) { +func (b *PollingBase) releaseConcurrency(_ context.Context, taskType string) { + ctx, cancel := pollingFallbackContext() + defer cancel() + currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType) if err := b.redis.Decr(ctx, currentKey).Err(); err != nil { - b.logger.Warn("释放并发计数失败", zap.Error(err)) + b.logger.Warn("释放并发计数失败", zap.String("task_type", taskType), zap.Error(err)) } } @@ -106,24 +117,43 @@ func (b *PollingBase) acquireCardTrafficSyncLock(ctx context.Context, cardID uin } // releaseCardTrafficSyncLock 释放卡流量同步锁。 -func (b *PollingBase) releaseCardTrafficSyncLock(ctx context.Context, cardID uint) { +func (b *PollingBase) releaseCardTrafficSyncLock(_ context.Context, cardID uint) { if b.redis == nil { return } + ctx, cancel := pollingFallbackContext() + defer cancel() + if err := b.redis.Del(ctx, constants.RedisCardTrafficSyncLockKey(cardID)).Err(); err != nil { b.logger.Warn("释放卡流量同步锁失败", zap.Uint("card_id", cardID), zap.Error(err)) } } +// requeueCardAt 使用独立短超时上下文执行真正的 ZADD 重入队。 +func (b *PollingBase) requeueCardAt(cardID uint, taskType string, nextCheckAt time.Time) error { + ctx, cancel := pollingFallbackContext() + defer cancel() + + if err := b.queueMgr.Requeue(ctx, cardID, taskType, nextCheckAt); err != nil { + b.logger.Error("重入队失败，卡可能暂停轮询", + zap.Uint("card_id", cardID), zap.String("task_type", taskType), zap.Error(err)) + return err + } + return nil +} + // requeueCard 将卡按匹配配置间隔重新入队分片 Sorted Set // ⚠️ 关键：Lua 脚本原子出队后卡已从队列移除，若并发满时直接 return 会导致卡永久丢失 // 调用方必须在 acquireConcurrency 返回 false 时调用此方法入队后再返回 -func (b *PollingBase) requeueCard(ctx context.Context, cardID uint, taskType string) error { +func (b *PollingBase) requeueCard(_ context.Context, cardID uint, taskType string) error { + ctx, cancel := pollingFallbackContext() card, err := b.getCardWithCache(ctx, cardID) + cancel() if err != nil { - b.logger.Warn("重入队：获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(err)) + b.logger.Warn("重入队：获取卡信息失败，使用30秒兜底间隔", + zap.Uint("card_id", cardID), zap.String("task_type", taskType), zap.Error(err)) // 获取卡失败时立即重入队（下次仍然尝试处理） - return b.queueMgr.Requeue(ctx, cardID, taskType, time.Now().Add(30*time.Second)) + return b.requeueCardAt(cardID, taskType, time.Now().Add(30*time.Second)) } intervals := b.configMgr.MergedTaskIntervals(card) @@ -131,7 +161,7 @@ func (b *PollingBase) requeueCard(ctx context.Context, cardID uint, taskType str // 兜底：配置未加载时延迟 30 秒重入队，防止卡从轮询永久消失 b.logger.Warn("无匹配轮询配置，30秒后重入队（配置可能未加载）", zap.Uint("card_id", cardID), zap.String("task_type", taskType)) - return b.queueMgr.Requeue(ctx, cardID, taskType, time.Now().Add(30*time.Second)) + return b.requeueCardAt(cardID, taskType, time.Now().Add(30*time.Second)) } info, ok := intervals[taskType] if !ok || info.Interval <= 0 { @@ -139,5 +169,5 @@ func (b *PollingBase) requeueCard(ctx context.Context, cardID uint, taskType str return nil } nextCheckAt := time.Now().Add(time.Duration(info.Interval) * time.Second) - return b.queueMgr.Requeue(ctx, cardID, taskType, nextCheckAt) + return b.requeueCardAt(cardID, taskType, nextCheckAt) } diff --git a/internal/task/polling_realname_handler.go b/internal/task/polling_realname_handler.go index f67a8a2..30e949a 100644 --- a/internal/task/polling_realname_handler.go +++ b/internal/task/polling_realname_handler.go @@ -208,7 +208,7 @@ func (h *PollingRealnameHandler) triggerFirstRealnameActivation(ctx context.Cont } task := asynq.NewTask(constants.TaskTypePackageFirstActivation, payloadBytes, asynq.MaxRetry(3), - asynq.Timeout(30*time.Second), + asynq.Timeout(60*time.Second), asynq.Queue(constants.QueueForTaskType(constants.TaskTypePackageFirstActivation)), ) if _, err := h.asynqClient.EnqueueContext(ctx, task); err != nil { @@ -246,7 +246,7 @@ func (h *PollingRealnameHandler) triggerDeviceRealnameActivation(ctx context.Con } task := asynq.NewTask(constants.TaskTypePackageFirstActivation, payloadBytes, asynq.MaxRetry(3), - asynq.Timeout(30*time.Second), + asynq.Timeout(60*time.Second), asynq.Queue(constants.QueueForTaskType(constants.TaskTypePackageFirstActivation)), ) if _, err := h.asynqClient.EnqueueContext(ctx, task); err != nil { diff --git a/pkg/config/defaults/config.yaml b/pkg/config/defaults/config.yaml index 61e8fcd..2ea2990 100644 --- a/pkg/config/defaults/config.yaml +++ b/pkg/config/defaults/config.yaml @@ -122,7 +122,7 @@ gateway: base_url: "https://lplan.whjhft.com/openapi" app_id: "60bgt1X8i7AvXqkd" app_secret: "BZeQttaZQt0i73moF" - timeout: 30 + timeout: 60 # 轮询配置 polling: