Преглед на файлове

feat:站点信息同步支持取消,任务超时调为4小时

- 同步循环每页检查 ctx:取消时记录标记为 cancelled,超时按失败处理
- 新增 fwhQueue.CancelSiteInfoUpdate:执行中的通知 worker 停止,排队中的直接删除;已结束的不覆盖
- 同步任务超时从 60 分钟调为 4 小时(循环响应 ctx 后,原超时会中途停止同步)
- InitEnqueueClient 同时创建 Inspector,供 fwh-agent-mcp 取消同步
- 新增按 id 查询同步记录、Progress 新增 cancelled

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Jerry Liang преди 4 дни
родител
ревизия
9ee8b08211

+ 6 - 5
service/domain/taskQueueDomain.go

@@ -5,11 +5,12 @@ type Trigger string  // 'auto' | 'manual'
 
 // Progress枚举
 const (
-	ProgressPending  Progress = "pending"
-	ProgressRunning  Progress = "running"
-	ProgressFailed   Progress = "failed"
-	ProgressError    Progress = "error"
-	ProgressComplete Progress = "complete"
+	ProgressPending   Progress = "pending"
+	ProgressRunning   Progress = "running"
+	ProgressFailed    Progress = "failed"
+	ProgressError     Progress = "error"
+	ProgressComplete  Progress = "complete"
+	ProgressCancelled Progress = "cancelled" // 用户取消(目前用于站点信息同步)
 )
 
 // Trigger枚举

+ 18 - 0
service/repositories/mongo/siteInfoManageRepo.go

@@ -96,6 +96,24 @@ func GetSiteInfoUpdateNewRecord() (*domain.SiteInfoUpdateRecord, error) {
 	return &result, nil
 }
 
+// 按 id 获取站点信息更新记录;不存在时返回 nil
+func GetSiteInfoUpdateRecord(id primitive.ObjectID) (*domain.SiteInfoUpdateRecord, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	collection := database.GetCollection(CollectionName_SiteInfoManageRecord)
+
+	var result domain.SiteInfoUpdateRecord
+	err := collection.FindOne(ctx, bson.M{"_id": id}).Decode(&result)
+	if err != nil {
+		if errors.Is(err, mongo.ErrNoDocuments) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return &result, nil
+}
+
 // 获取最近一次成功完成的站点信息更新记录;没有时返回 nil
 func GetSiteInfoUpdateLastComplete() (*domain.SiteInfoUpdateRecord, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

+ 1 - 0
service/repositories/repositories.go

@@ -423,6 +423,7 @@ var CreateSiteInfoUpdateInfo = mongo.CreateSiteInfoUpdateInfo
 var UpdateSiteInfoUpdateResult = mongo.UpdateSiteInfoUpdateResult
 var GetSiteInfoUpdateNewRecord = mongo.GetSiteInfoUpdateNewRecord
 var GetSiteInfoUpdateLastComplete = mongo.GetSiteInfoUpdateLastComplete
+var GetSiteInfoUpdateRecord = mongo.GetSiteInfoUpdateRecord
 var GetSiteInfoManageList = mongo.GetSiteInfoManageList
 var InsertSiteInfoData = mongo.InsertSiteInfoData
 var GetSiteInfoManageListFromGiccid = mongo.GetSiteInfoManageListFromGiccid

+ 2 - 1
service/task/queue/fwhQueue/fwhEnqueue.go

@@ -120,7 +120,8 @@ func EnqueueSiteInfoUpdateTask(payload domain.TaskSiteInfoUpdate) error {
 	taskId := newTaskID(TaskTypeSiteInfoUpdate, payload.Id.Hex())
 	task := asynq.NewTask(TaskTypeSiteInfoUpdate, data,
 		asynq.MaxRetry(0),
-		asynq.Timeout(60*time.Minute),
+		// 全量同步 6 万多个站点约 1 小时,可能超过 1.5 小时;同步循环会响应 ctx,超时即停止,所以留足余量
+		asynq.Timeout(SiteInfoUpdateTimeout),
 		asynq.Queue(QueueMain),
 		asynq.TaskID(taskId),
 		asynq.Unique(30*time.Minute), // 30 分钟内去重(保证唯一性)

+ 2 - 1
service/task/queue/fwhQueue/fwhTaskQueue.go

@@ -33,10 +33,11 @@ var (
 	inspector  *asynq.Inspector
 )
 
-// InitEnqueueClient 只创建入队客户端,不启动 worker、不注册处理函数。
+// InitEnqueueClient 只创建入队客户端与 Inspector(查询、取消),不启动 worker、不注册处理函数。
 // 供 fwh-agent-mcp 等独立进程把任务投递给主服务执行;主服务使用 InitFwhTaskQueue。
 func InitEnqueueClient(redisOpt asynq.RedisClientOpt) {
 	TaskClient = asynq.NewClient(redisOpt)
+	inspector = asynq.NewInspector(redisOpt)
 }
 
 // InitFwhTaskQueue 初始化新的 FWH 任务队列

+ 43 - 0
service/task/queue/fwhQueue/fwhTaskSiteInfoUpdate.go

@@ -3,12 +3,15 @@ package fwhQueue
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"fwh-toolbox-server/domain"
 	"fwh-toolbox-server/dto"
 	"fwh-toolbox-server/repositories"
+	"time"
 
 	"github.com/hibiken/asynq"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 	"go.uber.org/zap"
 )
 
@@ -29,6 +32,12 @@ func handleSiteInfoUpdate(ctx context.Context, task *asynq.Task) error {
 
 	// 执行任务
 	total, err := executeSiteInfoUpdate(ctx, payload)
+	if errors.Is(err, context.Canceled) {
+		// CancelSiteInfoUpdate 发出的取消(asynq 超时为 DeadlineExceeded,按失败处理)
+		zap.L().Info("[SiteInfoUpdate] 任务已取消", zap.String("id", taskId))
+		repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressCancelled, 0)
+		return err
+	}
 	if err != nil {
 		zap.L().Error("[SiteInfoUpdate] 任务执行失败", zap.String("id", taskId), zap.String("err", err.Error()))
 		repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressError, 0)
@@ -48,6 +57,10 @@ func executeSiteInfoUpdate(ctx context.Context, payload domain.TaskSiteInfoUpdat
 	var realTotal uint64 = 0
 
 	for {
+		// 每页检查一次:取消(CancelSiteInfoUpdate)或超时(SiteInfoUpdateTimeout)时停止
+		if err := ctx.Err(); err != nil {
+			return 0, err
+		}
 		resp, err := repositories.QuerySiteInfoList(dto.SiteInfoPara{
 			PageSize: pageSize,
 			PageNum:  pageNum,
@@ -72,3 +85,33 @@ func executeSiteInfoUpdate(ctx context.Context, payload domain.TaskSiteInfoUpdat
 
 	return realTotal, nil
 }
+
+// SiteInfoUpdateTimeout 站点信息同步任务的执行超时
+const SiteInfoUpdateTimeout = 4 * time.Hour
+
+// CancelSiteInfoUpdate 取消一次站点信息同步:
+// 正在执行的通知 worker 停止(worker 停下后把记录标记为 cancelled);还在排队或已不在队列中的直接删除并标记。
+// 可在主服务或只入队的独立进程(fwh-agent-mcp)中调用,需要 inspector 已初始化。
+func CancelSiteInfoUpdate(id primitive.ObjectID) error {
+	if inspector == nil {
+		return fmt.Errorf("任务队列未初始化")
+	}
+	taskId := newTaskID(TaskTypeSiteInfoUpdate, id.Hex())
+	info, err := inspector.GetTaskInfo(QueueMain, taskId)
+	switch {
+	case errors.Is(err, asynq.ErrTaskNotFound) || errors.Is(err, asynq.ErrQueueNotFound):
+		// 队列中已没有这个任务(如主服务重启时丢失):只更新记录
+	case err != nil:
+		return err
+	case info.State == asynq.TaskStateActive:
+		return inspector.CancelProcessing(taskId)
+	case info.State == asynq.TaskStateCompleted || info.State == asynq.TaskStateArchived:
+		// 任务已经结束,worker 已写入最终状态,不覆盖
+		return nil
+	case info.State == asynq.TaskStatePending || info.State == asynq.TaskStateScheduled || info.State == asynq.TaskStateRetry:
+		if err := inspector.DeleteTask(QueueMain, taskId); err != nil {
+			return err
+		}
+	}
+	return repositories.UpdateSiteInfoUpdateResult(id, domain.ProgressCancelled, 0)
+}