Merge branch 'main' of https://git.boss160.cn/csxj2026/junhong_cmp_fiber

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

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)),
 	)
 

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 {

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)
 }

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 {

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: