fix(task): tasks keep being cancelled (#1745)

* fix_cancel

* update(go.mod): update tache version

* tache v0.2.2

---------

Co-authored-by: j2rong4cn <j2rong@qq.com>

Author: jenfonro

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/KirCute/zip v1.0.1
 	github.com/OpenListTeam/go-cache v0.1.0
 	github.com/OpenListTeam/sftpd-openlist v1.0.1
-	github.com/OpenListTeam/tache v0.2.1
+	github.com/OpenListTeam/tache v0.2.2
 	github.com/OpenListTeam/times v0.1.0
 	github.com/OpenListTeam/wopan-sdk-go v0.1.5
 	github.com/ProtonMail/go-crypto v1.3.0

go.sum

@@ -55,8 +55,8 @@ github.com/OpenListTeam/gsync v0.1.0 h1:ywzGybOvA3lW8K1BUjKZ2IUlT2FSlzPO4DOazfYX
 github.com/OpenListTeam/gsync v0.1.0/go.mod h1:h/Rvv9aX/6CdW/7B8di3xK3xNV8dUg45Fehrd/ksZ9s=
 github.com/OpenListTeam/sftpd-openlist v1.0.1 h1:j4S3iPFOpnXCUKRPS7uCT4mF2VCl34GyqvH6lqwnkUU=
 github.com/OpenListTeam/sftpd-openlist v1.0.1/go.mod h1:uO/wKnbvbdq3rBLmClMTZXuCnw7XW4wlAq4dZe91a40=
-github.com/OpenListTeam/tache v0.2.1 h1:Uy/xAr05clHuMrr9+5fXAhv0Z5PGJivp4P5DnRez6cw=
-github.com/OpenListTeam/tache v0.2.1/go.mod h1:qmnZ/VpY2DUlmjg3UoDeNFy/LRqrw0biN3hYEEGc/+A=
+github.com/OpenListTeam/tache v0.2.2 h1:CWFn6sr1AIYaEjC8ONdKs+LrxHyuErheenAjEqRhh4k=
+github.com/OpenListTeam/tache v0.2.2/go.mod h1:qmnZ/VpY2DUlmjg3UoDeNFy/LRqrw0biN3hYEEGc/+A=
 github.com/OpenListTeam/times v0.1.0 h1:qknxw+qj5CYKgXAwydA102UEpPcpU8TYNGRmwRyPYpg=
 github.com/OpenListTeam/times v0.1.0/go.mod h1:Jx7qen5NCYzKk2w14YuvU48YYMcPa1P9a+EJePC15Pc=
 github.com/OpenListTeam/wopan-sdk-go v0.1.5 h1:iKKcVzIqBgtGDbn0QbdWrCazSGxXFmYFyrnFBG+U8dI=

internal/fs/archive.go

@@ -38,9 +38,6 @@ func (t *ArchiveDownloadTask) GetName() string {
 }
 
 func (t *ArchiveDownloadTask) Run() error {
-	if err := t.ReinitCtx(); err != nil {
-		return err
-	}
 	if t.SrcStorage == nil {
 		if srcStorage, _, err := op.GetStorageAndActualPath(t.SrcStorageMp); err == nil {
 			t.SrcStorage = srcStorage
@@ -155,9 +152,6 @@ func (t *ArchiveContentUploadTask) GetStatus() string {
 }
 
 func (t *ArchiveContentUploadTask) Run() error {
-	if err := t.ReinitCtx(); err != nil {
-		return err
-	}
 	t.ClearEndTime()
 	t.SetStartTime(time.Now())
 	defer func() { t.SetEndTime(time.Now()) }()

internal/fs/copy_move.go

@@ -13,7 +13,6 @@ import (
 	"github.com/OpenListTeam/OpenList/v4/internal/stream"
 	"github.com/OpenListTeam/OpenList/v4/internal/task"
 	"github.com/OpenListTeam/OpenList/v4/internal/task_group"
-	"github.com/OpenListTeam/OpenList/v4/pkg/utils"
 	"github.com/OpenListTeam/OpenList/v4/server/common"
 	"github.com/OpenListTeam/tache"
 	"github.com/pkg/errors"
@@ -48,9 +47,6 @@ func (t *FileTransferTask) GetName() string {
 }
 
 func (t *FileTransferTask) Run() error {
-	if err := t.ReinitCtx(); err != nil {
-		return err
-	}
 	if t.SrcStorage == nil {
 		if srcStorage, _, err := op.GetStorageAndActualPath(t.SrcStorageMp); err == nil {
 			t.SrcStorage = srcStorage
@@ -88,7 +84,7 @@ func (t *FileTransferTask) OnFailed() {
 }
 
 func (t *FileTransferTask) SetRetry(retry int, maxRetry int) {
-	t.TaskExtension.SetRetry(retry, maxRetry)
+	t.TaskData.SetRetry(retry, maxRetry)
 	if retry == 0 &&
 		(len(t.groupID) == 0 || // 重启恢复
 			(t.GetErr() == nil && t.GetState() != tache.StatePending)) { // 手动重试
@@ -198,17 +194,23 @@ func (t *FileTransferTask) RunWithNextTaskCallback(f func(nextTask *FileTransfer
 
 		existedObjs := make(map[string]bool)
 		if t.TaskType == merge {
-			dstObjs, _ := op.List(t.Ctx(), t.DstStorage, dstActualPath, model.ListArgs{})
+			dstObjs, err := op.List(t.Ctx(), t.DstStorage, dstActualPath, model.ListArgs{})
+			if err != nil {
+				return errors.WithMessagef(err, "failed list dst [%s] objs", dstActualPath)
+			}
 			for _, obj := range dstObjs {
+				if err := t.Ctx().Err(); err != nil {
+					return err
+				}
 				if !obj.IsDir() {
 					existedObjs[obj.GetName()] = true
 				}
 			}
 		}
 
 		for _, obj := range objs {
-			if utils.IsCanceled(t.Ctx()) {
-				return nil
+			if err := t.Ctx().Err(); err != nil {
+				return err
 			}
 
 			if t.TaskType == merge && !obj.IsDir() && existedObjs[obj.GetName()] {
@@ -239,6 +241,7 @@ func (t *FileTransferTask) RunWithNextTaskCallback(f func(nextTask *FileTransfer
 		return nil
 	}
 
+	t.Status = "getting src object link"
 	link, _, err := op.Link(t.Ctx(), t.SrcStorage, t.SrcActualPath, model.LinkArgs{})
 	if err != nil {
 		return errors.WithMessagef(err, "failed get [%s] link", t.SrcActualPath)

internal/offline_download/tool/download.go

@@ -32,9 +32,6 @@ type DownloadTask struct {
 }
 
 func (t *DownloadTask) Run() error {
-	if err := t.ReinitCtx(); err != nil {
-		return err
-	}
 	t.ClearEndTime()
 	t.SetStartTime(time.Now())
 	defer func() { t.SetEndTime(time.Now()) }()

internal/offline_download/tool/transfer.go

@@ -31,9 +31,6 @@ type TransferTask struct {
 }
 
 func (t *TransferTask) Run() error {
-	if err := t.ReinitCtx(); err != nil {
-		return err
-	}
 	if t.SrcStorage == nil && t.SrcStorageMp != "" {
 		if srcStorage, _, err := op.GetStorageAndActualPath(t.SrcStorageMp); err == nil {
 			t.SrcStorage = srcStorage
@@ -118,7 +115,7 @@ func (t *TransferTask) SetRetry(retry int, maxRetry int) {
 		t.groupID = stdpath.Join(t.DstStorageMp, t.DstActualPath)
 		task_group.TransferCoordinator.AddTask(t.groupID, nil)
 	}
-	t.TaskExtension.SetRetry(retry, maxRetry)
+	t.TaskData.SetRetry(retry, maxRetry)
 }
 
 var (

internal/task/base.go

@@ -65,18 +65,21 @@ func (t *TaskExtension) GetTotalBytes() int64 {
 	return t.TotalBytes
 }
 
-func (t *TaskExtension) ReinitCtx() error {
+func (t *TaskExtension) SetRetry(retry int, maxRetry int) {
+	t.Base.SetRetry(retry, maxRetry)
+	if retry > 0 || t.Ctx() == nil {
+		return
+	}
 	select {
 	case <-t.Ctx().Done():
 		if !conf.Conf.Tasks.AllowRetryCanceled {
-			return t.Ctx().Err()
+			return
 		}
 		ctx, cancel := context.WithCancel(context.Background())
 		t.SetCtx(ctx)
 		t.SetCancelFunc(cancel)
 	default:
 	}
-	return nil
 }
 
 type TaskExtensionInfo interface {