[fix][dingospeed] Task state repair in stop

Author: zhaoxianhua

internal/service/cache_job_service.go

@@ -40,18 +40,19 @@ func (p *CacheJobService) CreateCacheJob(c echo.Context, jobReq *query.CreateCac
 	ctx, cancelFunc := context.WithCancel(appInfo.Ctx())
 	var task common.Task
 	cacheTask := task2.CacheTask{
-		Ctx:          ctx,
-		Job:          jobReq,
-		CancelFunc:   cancelFunc,
-		SchedulerDao: p.schedulerDao,
+		Ctx:           ctx,
+		Job:           jobReq,
+		CancelFunc:    cancelFunc,
+		SchedulerDao:  p.schedulerDao,
+		RunningStatus: consts.RunningStatusJobBreak,
 	}
 	req := &manager.CreateCacheJobReq{
 		Type:       jobReq.Type,
 		InstanceId: jobReq.InstanceId,
 		Datatype:   jobReq.Datatype,
 		Org:        jobReq.Org,
 		Repo:       jobReq.Repo,
-		Status:     consts.StatusCacheJobIng,
+		Status:     consts.RunningStatusJobIng,
 	}
 	authorization := c.Request().Header.Get("Authorization")
 	if jobReq.Type == consts.CacheTypePreheat {
@@ -81,7 +82,7 @@ func (p *CacheJobService) CreateCacheJob(c echo.Context, jobReq *query.CreateCac
 			UsedStorage:   uint64(sha.UsedStorage),
 		}
 		if err = p.cachePool.SubmitForTimeout(ctx, task); err != nil {
-			p.schedulerDao.ExecUpdateCacheJobStatus(int(cacheJob.Id), consts.StatusCacheJobWait, jobReq.InstanceId, "", "", consts.TaskMoreErrMsg, 0)
+			p.schedulerDao.ExecUpdateCacheJobStatus(int(cacheJob.Id), consts.RunningStatusJobWait, jobReq.InstanceId, "", "", consts.TaskMoreErrMsg, 0)
 			return 0, err
 		}
 	} else if jobReq.Type == consts.CacheTypeMount {
@@ -91,7 +92,7 @@ func (p *CacheJobService) CreateCacheJob(c echo.Context, jobReq *query.CreateCac
 			Authorization: authorization,
 		}
 		if err := p.cachePool.SubmitForTimeout(ctx, task); err != nil {
-			p.schedulerDao.ExecUpdateRepositoryMountStatus(cacheTask.TaskNo, consts.StatusCacheJobWait, consts.TaskMoreErrMsg)
+			p.schedulerDao.ExecUpdateRepositoryMountStatus(cacheTask.TaskNo, consts.RunningStatusJobWait, consts.TaskMoreErrMsg)
 			return 0, err
 		}
 	} else {
@@ -103,10 +104,15 @@ func (p *CacheJobService) CreateCacheJob(c echo.Context, jobReq *query.CreateCac
 
 func (p *CacheJobService) StopCacheJob(jobStatusReq *query.JobStatusReq) error {
 	if task, ok := p.cachePool.GetTask(int(jobStatusReq.Id)); ok {
+		if t, ok := task.(*task2.PreheatCacheTask); ok {
+			t.RunningStatus = consts.RunningStatusJobStop
+		} else if t, ok := task.(*task2.MountCacheTask); ok {
+			t.RunningStatus = consts.RunningStatusJobStop
+		}
 		cancelFun := task.GetCancelFun()
 		cancelFun()
 	} else {
-		p.schedulerDao.ExecUpdateCacheJobStatus(int(jobStatusReq.Id), consts.StatusCacheJobBreak, jobStatusReq.InstanceId, "", "", "speed未注册该任务，下载已中断。", 0)
+		p.schedulerDao.ExecUpdateCacheJobStatus(int(jobStatusReq.Id), consts.RunningStatusJobStop, jobStatusReq.InstanceId, "", "", "speed未注册该任务，下载已中断。", 0)
 	}
 	return nil
 }
@@ -139,7 +145,7 @@ func (p *CacheJobService) ResumeCacheJob(c echo.Context, resumeCacheJobReq *quer
 			return err
 		}
 		// 将状态重置为进行中
-		p.schedulerDao.ExecUpdateCacheJobStatus(int(resumeCacheJobReq.Id), consts.StatusCacheJobIng,
+		p.schedulerDao.ExecUpdateCacheJobStatus(int(resumeCacheJobReq.Id), consts.RunningStatusJobIng,
 			resumeCacheJobReq.InstanceId, resumeCacheJobReq.Org, resumeCacheJobReq.Repo, "", 0)
 		task = &task2.PreheatCacheTask{
 			CacheTask:     cacheTask,
@@ -150,7 +156,7 @@ func (p *CacheJobService) ResumeCacheJob(c echo.Context, resumeCacheJobReq *quer
 			UsedStorage:   uint64(resumeCacheJobReq.UsedStorage),
 		}
 		if err := p.cachePool.SubmitForTimeout(ctx, task); err != nil {
-			p.schedulerDao.ExecUpdateCacheJobStatus(int(resumeCacheJobReq.Id), consts.StatusCacheJobWait, resumeCacheJobReq.InstanceId, "", "", consts.TaskMoreErrMsg, 0)
+			p.schedulerDao.ExecUpdateCacheJobStatus(int(resumeCacheJobReq.Id), consts.RunningStatusJobWait, resumeCacheJobReq.InstanceId, "", "", consts.TaskMoreErrMsg, 0)
 			return err
 		}
 	}

internal/service/task/mount_cache_task.go

@@ -75,10 +75,10 @@ func (m *MountCacheTask) DoTask() {
 				errMsg = strings.Join(lines, "\n")
 			}
 		}
-		m.SchedulerDao.ExecUpdateRepositoryMountStatus(m.TaskNo, consts.StatusCacheJobBreak, errMsg)
+		m.SchedulerDao.ExecUpdateRepositoryMountStatus(m.TaskNo, m.RunningStatus, errMsg)
 	} else {
 		zap.S().Infof("command success.%d", m.TaskNo)
-		m.SchedulerDao.ExecUpdateRepositoryMountStatus(m.TaskNo, consts.StatusCacheJobComplete, "")
+		m.SchedulerDao.ExecUpdateRepositoryMountStatus(m.TaskNo, consts.RunningStatusJobComplete, "")
 	}
 }
 

internal/service/task/preheat_cache_task.go

@@ -19,11 +19,12 @@ import (
 )
 
 type CacheTask struct {
-	TaskNo       int
-	Ctx          context.Context
-	CancelFunc   context.CancelFunc
-	Job          *query.CreateCacheJobReq
-	SchedulerDao *dao.SchedulerDao
+	TaskNo        int
+	Ctx           context.Context
+	CancelFunc    context.CancelFunc
+	Job           *query.CreateCacheJobReq
+	SchedulerDao  *dao.SchedulerDao
+	RunningStatus int32
 }
 
 func (c *CacheTask) GetTaskNo() int {
@@ -53,11 +54,11 @@ func (p *PreheatCacheTask) DoTask() {
 	go p.realTimeSpeed(ctx)
 	err := p.preheatProcess(orgRepo)
 	if err != nil {
-		p.SchedulerDao.ExecUpdateCacheJobStatus(p.TaskNo, consts.StatusCacheJobBreak, p.Job.InstanceId, p.Job.Org, p.Job.Repo, err.Error(), p.StockProcess)
+		p.SchedulerDao.ExecUpdateCacheJobStatus(p.TaskNo, p.RunningStatus, p.Job.InstanceId, p.Job.Org, p.Job.Repo, err.Error(), p.StockProcess)
 		return
 	}
 	p.StockProcess = 100
-	p.SchedulerDao.ExecUpdateCacheJobStatus(p.TaskNo, consts.StatusCacheJobComplete, p.Job.InstanceId, p.Job.Org, p.Job.Repo, "", p.StockProcess)
+	p.SchedulerDao.ExecUpdateCacheJobStatus(p.TaskNo, consts.RunningStatusJobComplete, p.Job.InstanceId, p.Job.Org, p.Job.Repo, "", p.StockProcess)
 }
 
 func (p *PreheatCacheTask) preheatProcess(orgRepo string) error {

pkg/consts/const.go

@@ -80,11 +80,12 @@ const (
 	CacheTypePreheat = 1
 	CacheTypeMount   = 2
 
-	StatusCacheJobIng      = 1
-	StatusCacheJobBreak    = 2
-	StatusCacheJobComplete = 3
-	StatusCacheJobStopping = 4
-	StatusCacheJobWait     = 5
+	RunningStatusJobIng      = 1
+	RunningStatusJobBreak    = 2
+	RunningStatusJobComplete = 3
+	RunningStatusJobStopping = 4
+	RunningStatusJobStop     = 5
+	RunningStatusJobWait     = 6
 
 	OperationProcess int = 1
 	OperationPreheat int = 2