fix: Fix file loss issues when copying or moving large directories

agent/app/api/v2/file.go

@@ -638,6 +638,26 @@ func (b *BaseApi) MoveFile(c *gin.Context) {
 	helper.Success(c)
 }
 
+// @Tags File
+// @Summary Stop file move task
+// @Accept json
+// @Param request body request.FileMoveStopReq true "request"
+// @Success 200
+// @Security ApiKeyAuth
+// @Security Timestamp
+// @Router /files/move/stop [post]
+func (b *BaseApi) StopMoveFile(c *gin.Context) {
+	var req request.FileMoveStopReq
+	if err := helper.CheckBindAndValidate(&req, c); err != nil {
+		return
+	}
+	if err := fileService.StopMvFile(req.TaskID); err != nil {
+		helper.InternalServer(c, err)
+		return
+	}
+	helper.Success(c)
+}
+
 // @Tags File
 // @Summary Download file
 // @Accept json

agent/app/dto/request/file.go

@@ -131,6 +131,11 @@ type FileMove struct {
 	Name       string   `json:"name"`
 	Cover      bool     `json:"cover"`
 	CoverPaths []string `json:"coverPaths"`
+	TaskID     string   `json:"taskID"`
+}
+
+type FileMoveStopReq struct {
+	TaskID string `json:"taskID" validate:"required"`
 }
 
 type FileDownload struct {

agent/app/service/file.go

@@ -52,6 +52,8 @@ type FileService struct {
 
 const fileHistorySnapshotMaxSize = 10 * 1024 * 1024
 
+var fileTransferLocks = newFileTransferLocks()
+
 type IFileService interface {
 	GetFileList(op request.FileOption) (response.FileInfo, error)
 	SearchUploadWithPage(req request.SearchUploadWithPage) (int64, interface{}, error)
@@ -72,6 +74,7 @@ type IFileService interface {
 	ChangeName(req request.FileRename) error
 	Wget(w request.FileWget) (string, error)
 	MvFile(m request.FileMove) error
+	StopMvFile(taskID string) error
 	ChangeOwner(req request.FileRoleUpdate) error
 	ChangeMode(op request.FileCreate) error
 	BatchChangeModeAndOwner(op request.FileRoleReq) error
@@ -908,17 +911,62 @@ func (f *FileService) Wget(w request.FileWget) (string, error) {
 
 func (f *FileService) MvFile(m request.FileMove) error {
 	fo := files.NewFileOp()
+	if err := validateFileMove(fo, m); err != nil {
+		return err
+	}
+	if m.TaskID == "" {
+		m.TaskID = common.GetUuid()
+	}
+	if !fileTransferLocks.Acquire(m.TaskID, getFileTransferPaths(m)) {
+		return buserr.New("TaskIsExecuting")
+	}
+	taskItem, err := task.NewTask(m.NewPath, task.TaskExec, task.TaskScopeTask, m.TaskID, 1)
+	if err != nil {
+		fileTransferLocks.Release(m.TaskID)
+		return err
+	}
+	go func() {
+		defer fileTransferLocks.Release(m.TaskID)
+		taskItem.AddSubTaskWithOps(m.NewPath, func(t *task.Task) error {
+			t.LogStart(m.NewPath)
+			err := f.moveFileWithContext(t.TaskCtx, m)
+			if err != nil && t.TaskCtx.Err() != nil {
+				return t.TaskCtx.Err()
+			}
+			return err
+		}, nil, 0, 0)
+		_ = taskItem.Execute()
+	}()
+	return nil
+}
+
+func (f *FileService) StopMvFile(taskID string) error {
+	if cancel, ok := global.LoadTaskCancel(taskID); ok {
+		cancel()
+		return nil
+	}
+	return buserr.New("TaskNotFound")
+}
+
+func validateFileMove(fo files.FileOp, m request.FileMove) error {
 	if !fo.Stat(m.NewPath) {
 		return buserr.New("ErrPathNotFound")
 	}
-	for _, oldPath := range m.OldPaths {
+	for _, oldPath := range append(append([]string{}, m.OldPaths...), m.CoverPaths...) {
 		if !fo.Stat(oldPath) {
 			return buserr.WithName("ErrFileNotFound", oldPath)
 		}
-		if oldPath == m.NewPath || strings.Contains(m.NewPath, filepath.Clean(oldPath)+"/") {
+		oldPath = filepath.Clean(oldPath)
+		newPath := filepath.Clean(m.NewPath)
+		if oldPath == newPath || strings.HasPrefix(newPath, oldPath+string(filepath.Separator)) {
 			return buserr.New("ErrMovePathFailed")
 		}
 	}
+	return nil
+}
+
+func (f *FileService) moveFileWithContext(ctx context.Context, m request.FileMove) error {
+	fo := files.NewFileOp()
 	type moveSnapshot struct {
 		path    string
 		content []byte
@@ -934,13 +982,25 @@ func (f *FileService) MvFile(m request.FileMove) error {
 		}
 		if len(m.CoverPaths) > 0 {
 			for _, src := range m.CoverPaths {
-				if err := fo.CopyAndReName(src, m.NewPath, "", true); err != nil {
+				if err := ctx.Err(); err != nil {
+					return err
+				}
+				if err := fo.CopyAndReNameWithContext(ctx, src, m.NewPath, "", true); err != nil {
 					errs = append(errs, err)
 					global.LOG.Errorf("cut copy file [%s] to [%s] failed, err: %s", src, m.NewPath, err.Error())
+					continue
+				}
+				if err := ctx.Err(); err != nil {
+					return err
+				}
+				if err := fo.DeleteDir(src); err != nil {
+					removeErr := fmt.Errorf("remove merged source [%s] failed: %w", src, err)
+					errs = append(errs, removeErr)
+					global.LOG.Errorf("%s", removeErr.Error())
 				}
 			}
 		}
-		if err := fo.Cut(m.OldPaths, m.NewPath, m.Name, m.Cover); err != nil {
+		if err := fo.CutWithContext(ctx, m.OldPaths, m.NewPath, m.Name, m.Cover); err != nil {
 			return err
 		}
 		for _, snapshot := range snapshots {
@@ -951,33 +1011,26 @@ func (f *FileService) MvFile(m request.FileMove) error {
 				}
 			}
 		}
-		return nil
+		return aggregateFileMoveErrors(errs)
 	}
 	if m.Type == "copy" {
 		for _, src := range m.OldPaths {
-			if err := fo.CopyAndReName(src, m.NewPath, m.Name, m.Cover); err != nil {
+			if err := fo.CopyAndReNameWithContext(ctx, src, m.NewPath, m.Name, m.Cover); err != nil {
 				errs = append(errs, err)
 				global.LOG.Errorf("copy file [%s] to [%s] failed, err: %s", src, m.NewPath, err.Error())
 			}
 		}
 		if len(m.CoverPaths) > 0 {
 			for _, src := range m.CoverPaths {
-				if err := fo.CopyAndReName(src, m.NewPath, "", true); err != nil {
+				if err := fo.CopyAndReNameWithContext(ctx, src, m.NewPath, "", true); err != nil {
 					errs = append(errs, err)
 					global.LOG.Errorf("copy file [%s] to [%s] failed, err: %s", src, m.NewPath, err.Error())
 				}
 			}
 		}
 	}
 
-	var errString string
-	for _, err := range errs {
-		errString += err.Error() + "\n"
-	}
-	if errString != "" {
-		return errors.New(errString)
-	}
-	return nil
+	return aggregateFileMoveErrors(errs)
 }
 
 func readEditableFileHistoryContent(filePath string) ([]byte, os.FileMode, bool) {

agent/app/service/file_transfer.go

@@ -0,0 +1,77 @@
+package service
+
+import (
+	"errors"
+	"path/filepath"
+	"strings"
+	"sync"
+
+	"github.com/1Panel-dev/1Panel/agent/app/dto/request"
+)
+
+type fileTransferLockSet struct {
+	mu    sync.Mutex
+	paths map[string][]string
+}
+
+func newFileTransferLocks() *fileTransferLockSet {
+	return &fileTransferLockSet{paths: make(map[string][]string)}
+}
+
+func (s *fileTransferLockSet) Acquire(taskID string, transferPaths []string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for _, activePaths := range s.paths {
+		for _, activePath := range activePaths {
+			for _, transferPath := range transferPaths {
+				if fileTransferPathsOverlap(activePath, transferPath) {
+					return false
+				}
+			}
+		}
+	}
+	s.paths[taskID] = transferPaths
+	return true
+}
+
+func (s *fileTransferLockSet) Release(taskID string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	delete(s.paths, taskID)
+}
+
+func getFileTransferPaths(req request.FileMove) []string {
+	paths := make([]string, 0, 1+len(req.OldPaths)+len(req.CoverPaths))
+	paths = append(paths, req.NewPath)
+	paths = append(paths, req.OldPaths...)
+	paths = append(paths, req.CoverPaths...)
+
+	unique := make(map[string]struct{}, len(paths))
+	result := make([]string, 0, len(paths))
+	for _, item := range paths {
+		item = filepath.Clean(item)
+		if _, ok := unique[item]; ok {
+			continue
+		}
+		unique[item] = struct{}{}
+		result = append(result, item)
+	}
+	return result
+}
+
+func fileTransferPathsOverlap(first, second string) bool {
+	return first == second || strings.HasPrefix(first, second+string(filepath.Separator)) || strings.HasPrefix(second, first+string(filepath.Separator))
+}
+
+func aggregateFileMoveErrors(errs []error) error {
+	if len(errs) == 0 {
+		return nil
+	}
+	var errString strings.Builder
+	for _, err := range errs {
+		errString.WriteString(err.Error())
+		errString.WriteByte('\n')
+	}
+	return errors.New(errString.String())
+}

agent/router/ro_file.go

@@ -44,6 +44,7 @@ func (f *FileRouter) InitRouter(Router *gin.RouterGroup) {
 		fileRouter.POST("/wget", baseApi.WgetFile)
 		fileRouter.POST("/wget/stop", baseApi.StopWget)
 		fileRouter.POST("/move", baseApi.MoveFile)
+		fileRouter.POST("/move/stop", baseApi.StopMoveFile)
 		fileRouter.GET("/download", baseApi.Download)
 		fileRouter.POST("/share/search", baseApi.SearchFileShare)
 		fileRouter.POST("/share/detail", baseApi.GetFileShareDetail)

agent/utils/files/file_op.go

@@ -594,6 +594,12 @@ func (f FileOp) DownloadFile(url, dst string) error {
 }
 
 func (f FileOp) Cut(oldPaths []string, dst, name string, cover bool) error {
+	ctx, cancel := context.WithTimeout(context.Background(), cmdRecursiveTimeout)
+	defer cancel()
+	return f.CutWithContext(ctx, oldPaths, dst, name, cover)
+}
+
+func (f FileOp) CutWithContext(ctx context.Context, oldPaths []string, dst, name string, cover bool) error {
 	if len(oldPaths) == 0 {
 		return nil
 	}
@@ -617,7 +623,7 @@ func (f FileOp) Cut(oldPaths []string, dst, name string, cover bool) error {
 	}
 	args = append(args, oldPaths...)
 	args = append(args, dstPath)
-	if err := cmd.NewCommandMgr(cmd.WithTimeout(cmdRecursiveTimeout)).Run("mv", args...); err != nil {
+	if err := cmd.NewCommandMgr(cmd.WithContext(ctx)).Run("mv", args...); err != nil {
 		return err
 	}
 	return nil

frontend/src/api/interface/file.ts

@@ -233,6 +233,11 @@ export namespace File {
         name?: string;
         cover?: boolean;
         coverPaths?: string[];
+        taskID?: string;
+    }
+
+    export interface FileMoveStopReq {
+        taskID: string;
     }
 
     export interface FileDownload {

frontend/src/api/modules/files.ts

@@ -146,7 +146,16 @@ export const stopWgetFile = (key: string) => {
 };
 
 export const moveFile = (params: File.FileMove) => {
-    return http.post<File.File>('files/move', params, TimeoutEnum.T_5M);
+    return http.post<File.File>('files/move', params);
+};
+
+export const stopMoveFile = (taskID: string, currentNode?: string) => {
+    return http.post(
+        'files/move/stop',
+        { taskID } as File.FileMoveStopReq,
+        undefined,
+        currentNode ? { CurrentNode: currentNode } : undefined,
+    );
 };
 
 export const downloadFile = (params: File.FileDownload) => {