From dedbba2db049d91d96ad9f48c947433e1342e5db Mon Sep 17 00:00:00 2001 From: ifloppy <68799904+ifloppy@users.noreply.github.com> Date: Sun, 26 Jul 2026 21:58:51 +0800 Subject: [PATCH 1/2] feat(task): add path-prefix batch cancel, delete and retry - Add `TaskWithPaths` interface and `MatchTaskPath` helper for matching tasks by virtual path prefix - Expose `GetSrcPath`/`GetDstPath` on transfer, upload, download and archive tasks - Add `delete_by_path`, `cancel_by_path` and `retry_by_path` endpoints for every task type - Cancel unfinished tasks before removing them so queued work stops - Scope matching to the requesting user and their base path - Exclude local temp and URL sources from path matching - Add unit and HTTP-level tests covering matching, permissions and large task sets Signed-off-by: ifloppy <68799904+ifloppy@users.noreply.github.com> --- internal/fs/archive.go | 11 + internal/fs/other.go | 18 ++ internal/fs/put.go | 11 + internal/fs/task_path_test.go | 50 ++++ internal/offline_download/tool/download.go | 8 + .../offline_download/tool/task_path_test.go | 33 +++ internal/task/path.go | 26 ++ internal/task/path_test.go | 31 +++ server/handles/task.go | 124 ++++++++- server/handles/task_path_http_test.go | 179 +++++++++++++ server/handles/task_path_test.go | 237 ++++++++++++++++++ 11 files changed, 727 insertions(+), 1 deletion(-) create mode 100644 internal/fs/task_path_test.go create mode 100644 internal/offline_download/tool/task_path_test.go create mode 100644 internal/task/path.go create mode 100644 internal/task/path_test.go create mode 100644 server/handles/task_path_http_test.go create mode 100644 server/handles/task_path_test.go diff --git a/internal/fs/archive.go b/internal/fs/archive.go index 784ba587a1..92b202aea8 100644 --- a/internal/fs/archive.go +++ b/internal/fs/archive.go @@ -147,6 +147,17 @@ func (t *ArchiveContentUploadTask) GetName() string { return fmt.Sprintf("upload %s to [%s](%s)", t.ObjName, t.DstStorageMp, t.DstActualPath) } +func (t *ArchiveContentUploadTask) GetSrcPath() string { + return "" +} + +func (t *ArchiveContentUploadTask) GetDstPath() string { + if t.DstStorageMp == "" { + return "" + } + return stdpath.Join(t.DstStorageMp, t.DstActualPath) +} + func (t *ArchiveContentUploadTask) GetStatus() string { return t.status } diff --git a/internal/fs/other.go b/internal/fs/other.go index a23beb73bc..d58888fce7 100644 --- a/internal/fs/other.go +++ b/internal/fs/other.go @@ -2,6 +2,7 @@ package fs import ( "context" + stdpath "path" "github.com/OpenListTeam/OpenList/v4/internal/conf" "github.com/OpenListTeam/OpenList/v4/internal/driver" @@ -62,3 +63,20 @@ type TaskData struct { func (t *TaskData) GetStatus() string { return t.Status } + +// GetSrcPath returns the virtual source path for task filtering. +// Local temp paths without a storage mount are excluded. +func (t *TaskData) GetSrcPath() string { + if t.SrcStorageMp == "" { + return "" + } + return stdpath.Join(t.SrcStorageMp, t.SrcActualPath) +} + +// GetDstPath returns the virtual destination path for task filtering. +func (t *TaskData) GetDstPath() string { + if t.DstStorageMp == "" { + return "" + } + return stdpath.Join(t.DstStorageMp, t.DstActualPath) +} diff --git a/internal/fs/put.go b/internal/fs/put.go index 0b905be08c..b0941ab072 100644 --- a/internal/fs/put.go +++ b/internal/fs/put.go @@ -31,6 +31,17 @@ func (t *UploadTask) GetName() string { return fmt.Sprintf("upload %s to [%s](%s)", t.file.GetName(), t.storage.GetStorage().MountPath, t.dstDirActualPath) } +func (t *UploadTask) GetSrcPath() string { + return "" +} + +func (t *UploadTask) GetDstPath() string { + if t.storage == nil { + return "" + } + return stdpath.Join(t.storage.GetStorage().MountPath, t.dstDirActualPath) +} + func (t *UploadTask) GetStatus() string { return "uploading" } diff --git a/internal/fs/task_path_test.go b/internal/fs/task_path_test.go new file mode 100644 index 0000000000..767c79881e --- /dev/null +++ b/internal/fs/task_path_test.go @@ -0,0 +1,50 @@ +package fs + +import "testing" + +func TestTaskDataGetPaths(t *testing.T) { + td := &TaskData{ + SrcActualPath: "src/file", + DstActualPath: "dst/dir", + SrcStorageMp: "/srcMount", + DstStorageMp: "/dstMount", + } + if got := td.GetSrcPath(); got != "/srcMount/src/file" { + t.Fatalf("GetSrcPath = %q", got) + } + if got := td.GetDstPath(); got != "/dstMount/dst/dir" { + t.Fatalf("GetDstPath = %q", got) + } + + localTemp := &TaskData{ + SrcActualPath: "/tmp/local-file", + DstActualPath: "dst", + DstStorageMp: "/cloud", + } + if localTemp.GetSrcPath() != "" { + t.Fatalf("local temp src must be empty for matching, got %q", localTemp.GetSrcPath()) + } + if localTemp.GetDstPath() != "/cloud/dst" { + t.Fatalf("GetDstPath = %q", localTemp.GetDstPath()) + } +} + +func TestArchiveContentUploadTaskGetPaths(t *testing.T) { + t1 := &ArchiveContentUploadTask{ + DstActualPath: "out", + DstStorageMp: "/mount", + } + if t1.GetSrcPath() != "" { + t.Fatal("src should be empty") + } + if t1.GetDstPath() != "/mount/out" { + t.Fatalf("dst = %q", t1.GetDstPath()) + } +} + +func TestUploadTaskGetPathsNilStorage(t *testing.T) { + t1 := &UploadTask{} + if t1.GetSrcPath() != "" || t1.GetDstPath() != "" { + t.Fatal("nil storage upload task paths should be empty") + } +} diff --git a/internal/offline_download/tool/download.go b/internal/offline_download/tool/download.go index b7f7a1a9df..a2535a9244 100644 --- a/internal/offline_download/tool/download.go +++ b/internal/offline_download/tool/download.go @@ -216,6 +216,14 @@ func (t *DownloadTask) GetName() string { return fmt.Sprintf("download %s to (%s)", t.Url, t.DstDirPath) } +func (t *DownloadTask) GetSrcPath() string { + return "" +} + +func (t *DownloadTask) GetDstPath() string { + return t.DstDirPath +} + func (t *DownloadTask) GetStatus() string { return t.Status } diff --git a/internal/offline_download/tool/task_path_test.go b/internal/offline_download/tool/task_path_test.go new file mode 100644 index 0000000000..cc8556bd3e --- /dev/null +++ b/internal/offline_download/tool/task_path_test.go @@ -0,0 +1,33 @@ +package tool + +import "testing" + +func TestDownloadTaskGetPaths(t *testing.T) { + tk := &DownloadTask{DstDirPath: "/cloud/dl"} + if tk.GetSrcPath() != "" { + t.Fatal("download src should be empty") + } + if tk.GetDstPath() != "/cloud/dl" { + t.Fatalf("dst = %q", tk.GetDstPath()) + } +} + +func TestTransferTaskGetPaths(t *testing.T) { + // TransferTask embeds fs.TaskData — local temp src excluded + tk := &TransferTask{} + tk.SrcActualPath = "/tmp/x" + tk.DstActualPath = "dir" + tk.DstStorageMp = "/dst" + if tk.GetSrcPath() != "" { + t.Fatalf("temp src must be empty, got %q", tk.GetSrcPath()) + } + if tk.GetDstPath() != "/dst/dir" { + t.Fatalf("dst = %q", tk.GetDstPath()) + } + + tk.SrcStorageMp = "/src" + tk.SrcActualPath = "a/b" + if tk.GetSrcPath() != "/src/a/b" { + t.Fatalf("src = %q", tk.GetSrcPath()) + } +} diff --git a/internal/task/path.go b/internal/task/path.go new file mode 100644 index 0000000000..bd1cee8447 --- /dev/null +++ b/internal/task/path.go @@ -0,0 +1,26 @@ +package task + +import ( + "github.com/OpenListTeam/OpenList/v4/pkg/utils" +) + +// TaskWithPaths is a task that exposes virtual src/dst paths for filtering. +// Empty string means the path side is not applicable for matching. +type TaskWithPaths interface { + TaskExtensionInfo + GetSrcPath() string + GetDstPath() string +} + +// MatchTaskPath reports whether src or dst is equal to prefix or under prefix. +// Empty src/dst sides are ignored. prefix/src/dst are cleaned before compare. +func MatchTaskPath(src, dst, prefix string) bool { + prefix = utils.FixAndCleanPath(prefix) + if src != "" && utils.IsSubPath(prefix, utils.FixAndCleanPath(src)) { + return true + } + if dst != "" && utils.IsSubPath(prefix, utils.FixAndCleanPath(dst)) { + return true + } + return false +} diff --git a/internal/task/path_test.go b/internal/task/path_test.go new file mode 100644 index 0000000000..db3fdd29df --- /dev/null +++ b/internal/task/path_test.go @@ -0,0 +1,31 @@ +package task + +import "testing" + +func TestMatchTaskPath(t *testing.T) { + tests := []struct { + name string + src string + dst string + prefix string + want bool + }{ + {name: "dst equal", src: "", dst: "/a/b", prefix: "/a/b", want: true}, + {name: "dst child", src: "", dst: "/a/b/c", prefix: "/a/b", want: true}, + {name: "src child", src: "/a/b/c", dst: "/other", prefix: "/a/b", want: true}, + {name: "no false prefix", src: "", dst: "/ab", prefix: "/a", want: false}, + {name: "unrelated", src: "/x", dst: "/y", prefix: "/a", want: false}, + {name: "empty sides", src: "", dst: "", prefix: "/a", want: false}, + {name: "root matches all", src: "/a", dst: "", prefix: "/", want: true}, + {name: "backslash cleaned", src: "", dst: "\\a\\b\\c", prefix: "/a/b", want: true}, + {name: "relative cleaned", src: "", dst: "a/b/c", prefix: "/a/b", want: true}, + {name: "only src empty ignored", src: "", dst: "/keep/me", prefix: "/other", want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := MatchTaskPath(tt.src, tt.dst, tt.prefix); got != tt.want { + t.Fatalf("MatchTaskPath(%q,%q,%q)=%v want %v", tt.src, tt.dst, tt.prefix, got, tt.want) + } + }) + } +} diff --git a/server/handles/task.go b/server/handles/task.go index 032f363ad0..d2070f35e9 100644 --- a/server/handles/task.go +++ b/server/handles/task.go @@ -1,7 +1,9 @@ package handles import ( + "errors" "math" + "strings" "time" "github.com/OpenListTeam/OpenList/v4/internal/conf" @@ -30,6 +32,19 @@ type TaskInfo struct { Error string `json:"error"` } +type TaskPathReq struct { + Path string `json:"path"` +} + +type TaskPathResult struct { + Count int `json:"count"` +} + +var errEmptyTaskPath = errors.New("path is required") + +// taskDoneStates are the terminal states of a task. +var taskDoneStates = []tache.State{tache.StateCanceled, tache.StateFailed, tache.StateSucceeded} + func getTaskInfo[T task.TaskExtensionInfo](task T) TaskInfo { errMsg := "" if task.GetErr() != nil { @@ -77,6 +92,36 @@ func getUserInfo(c *gin.Context) (bool, uint, bool) { } } +func getUser(c *gin.Context) (*model.User, bool) { + if user, ok := c.Request.Context().Value(conf.UserKey).(*model.User); ok { + return user, true + } + return nil, false +} + +func resolveTaskPathPrefix(user *model.User, raw string) (string, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return "", errEmptyTaskPath + } + if user.IsAdmin() { + return utils.FixAndCleanPath(raw), nil + } + return user.JoinPath(raw) +} + +func taskOwnedBy[T task.TaskExtensionInfo](t T, isAdmin bool, uid uint) bool { + if isAdmin { + return true + } + creator := t.GetCreator() + return creator != nil && creator.ID == uid +} + +func taskMatchesPathPrefix[T task.TaskWithPaths](t T, prefix string) bool { + return task.MatchTaskPath(t.GetSrcPath(), t.GetDstPath(), prefix) +} + func getTargetedHandler[T task.TaskExtensionInfo](manager task.Manager[T], callback func(c *gin.Context, task T)) gin.HandlerFunc { return func(c *gin.Context) { isAdmin, uid, ok := getUserInfo(c) @@ -124,7 +169,81 @@ func getBatchHandler[T task.TaskExtensionInfo](manager task.Manager[T], callback } } -func taskRoute[T task.TaskExtensionInfo](g *gin.RouterGroup, manager task.Manager[T]) { +type pathBatchOp[T task.TaskWithPaths] struct { + filter func(t T) bool + apply func(m task.Manager[T], t T) +} + +func applyPathBatch[T task.TaskWithPaths](manager task.Manager[T], isAdmin bool, uid uint, prefix string, op pathBatchOp[T]) int { + tasks := manager.GetByCondition(func(t T) bool { + return taskOwnedBy(t, isAdmin, uid) && taskMatchesPathPrefix(t, prefix) && op.filter(t) + }) + count := 0 + for _, t := range tasks { + op.apply(manager, t) + count++ + } + return count +} + +func getPathBatchHandler[T task.TaskWithPaths](manager task.Manager[T], op pathBatchOp[T]) gin.HandlerFunc { + return func(c *gin.Context) { + user, ok := getUser(c) + if !ok { + common.ErrorStrResp(c, "user invalid", 401) + return + } + var req TaskPathReq + if err := c.ShouldBind(&req); err != nil { + common.ErrorStrResp(c, "invalid request format", 400) + return + } + prefix, err := resolveTaskPathPrefix(user, req.Path) + if err != nil { + common.ErrorStrResp(c, err.Error(), 400) + return + } + count := applyPathBatch(manager, user.IsAdmin(), user.ID, prefix, op) + common.SuccessResp(c, TaskPathResult{Count: count}) + } +} + +func deleteByPath[T task.TaskWithPaths](manager task.Manager[T]) gin.HandlerFunc { + return getPathBatchHandler(manager, pathBatchOp[T]{ + filter: func(T) bool { return true }, + apply: func(m task.Manager[T], t T) { + // cancel first so running/pending work stops, then remove from manager + if !argsContains(t.GetState(), taskDoneStates...) { + m.Cancel(t.GetID()) + } + m.Remove(t.GetID()) + }, + }) +} + +func cancelByPath[T task.TaskWithPaths](manager task.Manager[T]) gin.HandlerFunc { + return getPathBatchHandler(manager, pathBatchOp[T]{ + filter: func(t T) bool { + return !argsContains(t.GetState(), taskDoneStates...) + }, + apply: func(m task.Manager[T], t T) { + m.Cancel(t.GetID()) + }, + }) +} + +func retryByPath[T task.TaskWithPaths](manager task.Manager[T]) gin.HandlerFunc { + return getPathBatchHandler(manager, pathBatchOp[T]{ + filter: func(t T) bool { + return t.GetState() == tache.StateFailed + }, + apply: func(m task.Manager[T], t T) { + m.Retry(t.GetID()) + }, + }) +} + +func taskRoute[T task.TaskWithPaths](g *gin.RouterGroup, manager task.Manager[T]) { g.GET("/undone", func(c *gin.Context) { isAdmin, uid, ok := getUserInfo(c) if !ok { @@ -215,6 +334,9 @@ func taskRoute[T task.TaskExtensionInfo](g *gin.RouterGroup, manager task.Manage } common.SuccessResp(c) }) + g.POST("/delete_by_path", deleteByPath(manager)) + g.POST("/cancel_by_path", cancelByPath(manager)) + g.POST("/retry_by_path", retryByPath(manager)) } func SetupTaskRoute(g *gin.RouterGroup) { diff --git a/server/handles/task_path_http_test.go b/server/handles/task_path_http_test.go new file mode 100644 index 0000000000..18453f2f9b --- /dev/null +++ b/server/handles/task_path_http_test.go @@ -0,0 +1,179 @@ +package handles + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/OpenListTeam/OpenList/v4/internal/conf" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/tache" + "github.com/gin-gonic/gin" +) + +func contextWithUser(ctx context.Context, user *model.User) context.Context { + return context.WithValue(ctx, conf.UserKey, user) +} + +// buildPathRouter wires the by-path handlers with an injected user, mimicking +// what middlewares.Auth does, so we can assert permission scoping end-to-end. +func buildPathRouter(m *tache.Manager[*pathTestTask], user *model.User) *gin.Engine { + gin.SetMode(gin.TestMode) + r := gin.New() + r.Use(func(c *gin.Context) { + if user != nil { + ctx := c.Request.Context() + //nolint:staticcheck // mirrors production context key usage + c.Request = c.Request.WithContext(contextWithUser(ctx, user)) + } + c.Next() + }) + g := r.Group("/task/copy") + g.POST("/delete_by_path", deleteByPath[*pathTestTask](m)) + g.POST("/cancel_by_path", cancelByPath[*pathTestTask](m)) + g.POST("/retry_by_path", retryByPath[*pathTestTask](m)) + return r +} + +func postPath(t *testing.T, r *gin.Engine, url, body string) (int, map[string]any) { + t.Helper() + req := httptest.NewRequest(http.MethodPost, url, strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + var out map[string]any + _ = json.Unmarshal(w.Body.Bytes(), &out) + return w.Code, out +} + +func respCode(out map[string]any) int { + if v, ok := out["code"].(float64); ok { + return int(v) + } + return -1 +} + +func respCount(t *testing.T, out map[string]any) int { + t.Helper() + data, ok := out["data"].(map[string]any) + if !ok { + t.Fatalf("no data object in response: %v", out) + } + c, ok := data["count"].(float64) + if !ok { + t.Fatalf("no count in data: %v", data) + } + return int(c) +} + +func TestDeleteByPathHTTPEmptyPathRejected(t *testing.T) { + ensureTaskTestConf() + admin := &model.User{ID: 1, Role: model.ADMIN} + m := newPathManager(1) + r := buildPathRouter(m, admin) + + _, out := postPath(t, r, "/task/copy/delete_by_path", `{"path":" "}`) + if got := respCode(out); got != 400 { + t.Fatalf("empty path code = %d, want 400 (%v)", got, out) + } +} + +func TestDeleteByPathHTTPNoUserRejected(t *testing.T) { + ensureTaskTestConf() + m := newPathManager(1) + r := buildPathRouter(m, nil) + + _, out := postPath(t, r, "/task/copy/delete_by_path", `{"path":"/x"}`) + if got := respCode(out); got != 401 { + t.Fatalf("missing user code = %d, want 401 (%v)", got, out) + } +} + +func TestDeleteByPathHTTPScopedToOwnerAndBasePath(t *testing.T) { + ensureTaskTestConf() + owner := &model.User{ID: 2, Role: model.GENERAL, BasePath: "/home/u2"} + stranger := &model.User{ID: 3, Role: model.GENERAL, BasePath: "/home/u3"} + m := newPathManager(1) + + // owner's task inside their base path -> should be deleted + mine := addPathTask(m, owner, "", "/home/u2/docs/a", tache.StatePending) + // another user's task under the same resolved path -> must survive + notMine := addPathTask(m, stranger, "", "/home/u2/docs/b", tache.StatePending) + // owner's task outside the requested prefix -> must survive + outside := addPathTask(m, owner, "", "/home/u2/other/c", tache.StatePending) + + r := buildPathRouter(m, owner) + // non-admin sends a relative path; it is joined onto their BasePath + _, out := postPath(t, r, "/task/copy/delete_by_path", `{"path":"docs"}`) + if got := respCode(out); got != 200 { + t.Fatalf("code = %d (%v)", got, out) + } + if got := respCount(t, out); got != 1 { + t.Fatalf("count = %d, want 1 (only own task under base path)", got) + } + if _, ok := m.GetByID(mine.GetID()); ok { + t.Fatal("owner task under prefix should be removed") + } + if _, ok := m.GetByID(notMine.GetID()); !ok { + t.Fatal("SECURITY: another user's task was removed") + } + if _, ok := m.GetByID(outside.GetID()); !ok { + t.Fatal("task outside prefix was removed") + } +} + +func TestDeleteByPathHTTPAdminNotEscapedByTraversal(t *testing.T) { + ensureTaskTestConf() + user := &model.User{ID: 4, Role: model.GENERAL, BasePath: "/home/u4"} + m := newPathManager(1) + + // a task that lives outside the user's base path + foreign := addPathTask(m, user, "", "/etc/secret/x", tache.StatePending) + + r := buildPathRouter(m, user) + // attempt to escape the base path via traversal + _, out := postPath(t, r, "/task/copy/delete_by_path", `{"path":"../../etc/secret"}`) + code := respCode(out) + if code == 200 { + if got := respCount(t, out); got != 0 { + t.Fatalf("SECURITY: traversal deleted %d task(s) outside base path", got) + } + } + if _, ok := m.GetByID(foreign.GetID()); !ok { + t.Fatal("SECURITY: task outside base path removed via path traversal") + } +} + +func TestCancelAndRetryByPathHTTPStateFiltering(t *testing.T) { + ensureTaskTestConf() + admin := &model.User{ID: 1, Role: model.ADMIN} + m := newPathManager(1) + + pending := addPathTask(m, admin, "", "/p/pending", tache.StatePending) + failed := addPathTask(m, admin, "", "/p/failed", tache.StateFailed) + done := addPathTask(m, admin, "", "/p/done", tache.StateSucceeded) + + r := buildPathRouter(m, admin) + + _, out := postPath(t, r, "/task/copy/cancel_by_path", `{"path":"/p"}`) + if got := respCount(t, out); got != 1 { + t.Fatalf("cancel count = %d, want 1 (pending only)", got) + } + if done.GetState() != tache.StateSucceeded { + t.Fatalf("succeeded task altered by cancel: %v", done.GetState()) + } + + _, out = postPath(t, r, "/task/copy/retry_by_path", `{"path":"/p"}`) + if got := respCount(t, out); got != 1 { + t.Fatalf("retry count = %d, want 1 (failed only)", got) + } + if failed.GetState() != tache.StateWaitingRetry { + t.Fatalf("failed task not queued for retry: %v", failed.GetState()) + } + if st := pending.GetState(); st != tache.StateCanceling && st != tache.StateCanceled { + t.Fatalf("pending task state after cancel = %v", st) + } +} diff --git a/server/handles/task_path_test.go b/server/handles/task_path_test.go new file mode 100644 index 0000000000..3e8def2b1f --- /dev/null +++ b/server/handles/task_path_test.go @@ -0,0 +1,237 @@ +package handles + +import ( + "fmt" + "testing" + "time" + + "github.com/OpenListTeam/OpenList/v4/internal/conf" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/internal/task" + "github.com/OpenListTeam/tache" +) + +func ensureTaskTestConf() { + if conf.Conf == nil { + conf.Conf = conf.DefaultConfig(".") + } +} + +type pathTestTask struct { + task.TaskExtension + src string + dst string +} + +func (t *pathTestTask) GetName() string { return "path-test" } +func (t *pathTestTask) GetStatus() string { return "" } +func (t *pathTestTask) Run() error { return nil } +func (t *pathTestTask) GetSrcPath() string { return t.src } +func (t *pathTestTask) GetDstPath() string { return t.dst } + +func newPathManager(workers int) *tache.Manager[*pathTestTask] { + return tache.NewManager[*pathTestTask]( + tache.WithWorks(workers), + tache.WithRunning(false), + ) +} + +func addPathTask(m *tache.Manager[*pathTestTask], creator *model.User, src, dst string, state tache.State) *pathTestTask { + t := &pathTestTask{ + TaskExtension: task.TaskExtension{Creator: creator}, + src: src, + dst: dst, + } + t.SetState(state) + m.Add(t) + return t +} + +func TestResolveTaskPathPrefix(t *testing.T) { + admin := &model.User{ID: 1, Role: model.ADMIN, BasePath: "/"} + user := &model.User{ID: 2, Role: model.GENERAL, BasePath: "/home/user"} + + got, err := resolveTaskPathPrefix(admin, " /Storage/a ") + if err != nil { + t.Fatal(err) + } + if got != "/Storage/a" { + t.Fatalf("admin path = %q", got) + } + + got, err = resolveTaskPathPrefix(user, "folder") + if err != nil { + t.Fatal(err) + } + if got != "/home/user/folder" { + t.Fatalf("user path = %q", got) + } + + if _, err := resolveTaskPathPrefix(admin, " "); err == nil { + t.Fatal("expected empty path error") + } +} + +func TestApplyPathBatchDeleteCancelRetry(t *testing.T) { + ensureTaskTestConf() + admin := &model.User{ID: 1, Role: model.ADMIN} + other := &model.User{ID: 2, Role: model.GENERAL} + m := newPathManager(1) + + keep := addPathTask(m, admin, "/keep/src", "/keep/dst", tache.StateSucceeded) + matchPending := addPathTask(m, admin, "/target/a", "/other", tache.StatePending) + matchDstDone := addPathTask(m, admin, "", "/target/b/c", tache.StateSucceeded) + matchFailed := addPathTask(m, admin, "", "/target/fail", tache.StateFailed) + otherUser := addPathTask(m, other, "", "/target/secret", tache.StatePending) + falsePrefix := addPathTask(m, admin, "", "/targetx/x", tache.StatePending) + + // non-admin can only touch own tasks + count := applyPathBatch(m, false, other.ID, "/target", pathBatchOp[*pathTestTask]{ + filter: func(*pathTestTask) bool { return true }, + apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { + if !argsContains(tk.GetState(), taskDoneStates...) { + manager.Cancel(tk.GetID()) + } + manager.Remove(tk.GetID()) + }, + }) + if count != 1 { + t.Fatalf("non-admin delete count = %d want 1", count) + } + if _, ok := m.GetByID(otherUser.GetID()); ok { + t.Fatal("other user matched task should be removed") + } + if _, ok := m.GetByID(matchPending.GetID()); !ok { + t.Fatal("admin task must remain for non-admin delete") + } + + // admin cancel by path: only unfinished matching + count = applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ + filter: func(tk *pathTestTask) bool { + return !argsContains(tk.GetState(), taskDoneStates...) + }, + apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { + manager.Cancel(tk.GetID()) + }, + }) + if count != 1 { + t.Fatalf("cancel count = %d want 1 (pending only; failed is done-state)", count) + } + // pending becomes canceling/canceled; failed stays failed; done stays + if st := matchPending.GetState(); st != tache.StateCanceling && st != tache.StateCanceled { + t.Fatalf("pending state after cancel = %v", st) + } + if matchFailed.GetState() != tache.StateFailed { + t.Fatalf("failed should not be canceled, got %v", matchFailed.GetState()) + } + if matchDstDone.GetState() != tache.StateSucceeded { + t.Fatalf("succeeded should stay, got %v", matchDstDone.GetState()) + } + + // retry by path: only failed + count = applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ + filter: func(tk *pathTestTask) bool { + return tk.GetState() == tache.StateFailed + }, + apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { + manager.Retry(tk.GetID()) + }, + }) + if count != 1 { + t.Fatalf("retry count = %d want 1", count) + } + if matchFailed.GetState() != tache.StateWaitingRetry { + t.Fatalf("failed after retry state = %v want waiting_retry", matchFailed.GetState()) + } + + // delete by path: cancel then remove all matching + count = applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ + filter: func(*pathTestTask) bool { return true }, + apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { + if !argsContains(tk.GetState(), taskDoneStates...) { + manager.Cancel(tk.GetID()) + } + manager.Remove(tk.GetID()) + }, + }) + // matchPending, matchDstDone, matchFailed still present under /target + if count != 3 { + t.Fatalf("delete count = %d want 3", count) + } + for _, id := range []string{matchPending.GetID(), matchDstDone.GetID(), matchFailed.GetID()} { + if _, ok := m.GetByID(id); ok { + t.Fatalf("task %s should be removed", id) + } + } + if _, ok := m.GetByID(keep.GetID()); !ok { + t.Fatal("unrelated task removed") + } + if _, ok := m.GetByID(falsePrefix.GetID()); !ok { + t.Fatal("false-prefix task removed") + } +} + +func TestApplyPathBatchLargeScale(t *testing.T) { + ensureTaskTestConf() + admin := &model.User{ID: 1, Role: model.ADMIN} + m := newPathManager(1) + + const total = 10000 + const matchN = 3000 + for i := 0; i < total; i++ { + dst := fmt.Sprintf("/other/%d", i) + state := tache.StateSucceeded + if i < matchN { + dst = fmt.Sprintf("/bulk/item/%d", i) + if i%3 == 0 { + state = tache.StatePending + } else if i%3 == 1 { + state = tache.StateFailed + } + } + addPathTask(m, admin, "", dst, state) + } + + start := time.Now() + count := applyPathBatch(m, true, admin.ID, "/bulk", pathBatchOp[*pathTestTask]{ + filter: func(*pathTestTask) bool { return true }, + apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { + if !argsContains(tk.GetState(), taskDoneStates...) { + manager.Cancel(tk.GetID()) + } + manager.Remove(tk.GetID()) + }, + }) + elapsed := time.Since(start) + if count != matchN { + t.Fatalf("count = %d want %d", count, matchN) + } + remain := len(m.GetAll()) + if remain != total-matchN { + t.Fatalf("remain = %d want %d", remain, total-matchN) + } + if elapsed > 5*time.Second { + t.Fatalf("delete %d of %d too slow: %s", matchN, total, elapsed) + } + t.Logf("deleted %d/%d tasks in %s", matchN, total, elapsed) +} + +func TestTaskOwnedBy(t *testing.T) { + admin := &model.User{ID: 1, Role: model.ADMIN} + user := &model.User{ID: 2, Role: model.GENERAL} + tk := &pathTestTask{TaskExtension: task.TaskExtension{Creator: user}} + + if !taskOwnedBy(tk, true, admin.ID) { + t.Fatal("admin should own all") + } + if !taskOwnedBy(tk, false, user.ID) { + t.Fatal("owner should own") + } + if taskOwnedBy(tk, false, admin.ID) { + t.Fatal("non-owner must not own") + } + nilCreator := &pathTestTask{} + if taskOwnedBy(nilCreator, false, user.ID) { + t.Fatal("nil creator must not match non-admin") + } +} From e8879b33ec15592075e3857a950da26a30895c26 Mon Sep 17 00:00:00 2001 From: ifloppy <68799904+ifloppy@users.noreply.github.com> Date: Mon, 27 Jul 2026 23:43:10 +0800 Subject: [PATCH 2/2] fix(task): tighten path batch operation contracts - Match upload and archive-content tasks by their final destination object path - Require an explicit slash for operations that resolve to the global root - Wait for running task cancellation and cleanup before removing tasks - Return separate matched and processed counts for concurrent state changes - Document normalization, scoping, state filters, lifecycle and response semantics - Cover exact-file matching, lifecycle cleanup, implicit root rejection and count races Signed-off-by: ifloppy <68799904+ifloppy@users.noreply.github.com> --- docs/task-path-batch-api.md | 92 +++++++++++++++ internal/fs/archive.go | 6 +- internal/fs/put.go | 4 +- internal/fs/task_path_test.go | 57 ++++++++- server/handles/task.go | 141 +++++++++++++++++----- server/handles/task_path_http_test.go | 42 +++++-- server/handles/task_path_test.go | 164 ++++++++++++++++++++------ 7 files changed, 420 insertions(+), 86 deletions(-) create mode 100644 docs/task-path-batch-api.md diff --git a/docs/task-path-batch-api.md b/docs/task-path-batch-api.md new file mode 100644 index 0000000000..85223350da --- /dev/null +++ b/docs/task-path-batch-api.md @@ -0,0 +1,92 @@ +# Task path-prefix batch API + +The task API provides batch operations for every task type registered under +`/api/task`: + +- `POST /api/task/{type}/delete_by_path` +- `POST /api/task/{type}/cancel_by_path` +- `POST /api/task/{type}/retry_by_path` + +Supported task types are `upload`, `copy`, `move`, `offline_download`, +`offline_download_transfer`, `decompress`, and `decompress_upload`. + +## Request + +```json +{ + "path": "/storage/folder" +} +``` + +`path` is normalized with the same virtual-path rules used by OpenList. Both +forward and backward slashes are accepted, repeated separators are collapsed, +and `.` / `..` segments are cleaned. + +A root-wide operation is destructive and therefore requires the explicit input +`/`. If the path after user-base-path resolution is `/`, inputs such as `.`, +`./`, `//`, or other non-empty values that normalize to `/` are rejected with +HTTP 400. + +For non-admin users, the normalized path is resolved against the user's base +path. A non-admin operation can only match tasks created by that user. An admin +can match tasks from all users. + +## Matching contract + +A task matches when either its virtual source path or virtual destination path: + +1. equals the normalized request path; or +2. is a descendant of the normalized request path. + +Matching is path-segment aware: `/a` matches `/a/file`, but not `/ab/file`. + +Upload tasks expose the final destination object path, including the uploaded +file name. Archive-content file and non-in-place directory tasks also include +their object name. An in-place archive directory task represents expansion into +the destination directory itself; its generated child tasks expose their final +object paths. + +Local temporary paths and download URLs are not virtual storage paths and are +not considered for source-path matching. + +## Operation behavior + +### `delete_by_path` + +Matches tasks in any state. Terminal tasks and queued tasks that have never +started are removed immediately. Active tasks are first cancelled, then the API +waits for execution to stop and failure cleanup hooks to finish before removing +them from the manager. A task that does not stop before the server-side timeout +remains visible in the manager and is not counted as processed. + +### `cancel_by_path` + +Matches only non-terminal tasks. Matching tasks are asked to cancel but remain +in the manager so their final state can be inspected. + +### `retry_by_path` + +Matches only tasks currently in the failed state. Matching tasks are queued for +retry. + +## Response + +```json +{ + "code": 200, + "message": "success", + "data": { + "matched": 125, + "processed": 124 + } +} +``` + +- `matched` is the number of tasks selected by the path, ownership, and state + filters at the start of the operation. +- `processed` is the number of those tasks on which the requested state-checked + operation was successfully applied. For deletion, it is the number confirmed + removed from the manager after any required cancellation and cleanup. + +Concurrent task transitions can make `processed` lower than `matched`. Clients +should display both values rather than treating `matched` as a success count. diff --git a/internal/fs/archive.go b/internal/fs/archive.go index 92b202aea8..fee5ee408c 100644 --- a/internal/fs/archive.go +++ b/internal/fs/archive.go @@ -155,7 +155,11 @@ func (t *ArchiveContentUploadTask) GetDstPath() string { if t.DstStorageMp == "" { return "" } - return stdpath.Join(t.DstStorageMp, t.DstActualPath) + path := stdpath.Join(t.DstStorageMp, t.DstActualPath) + if t.InPlace { + return path + } + return stdpath.Join(path, t.ObjName) } func (t *ArchiveContentUploadTask) GetStatus() string { diff --git a/internal/fs/put.go b/internal/fs/put.go index b0941ab072..8903023fd7 100644 --- a/internal/fs/put.go +++ b/internal/fs/put.go @@ -36,10 +36,10 @@ func (t *UploadTask) GetSrcPath() string { } func (t *UploadTask) GetDstPath() string { - if t.storage == nil { + if t.storage == nil || t.file == nil { return "" } - return stdpath.Join(t.storage.GetStorage().MountPath, t.dstDirActualPath) + return stdpath.Join(t.storage.GetStorage().MountPath, t.dstDirActualPath, t.file.GetName()) } func (t *UploadTask) GetStatus() string { diff --git a/internal/fs/task_path_test.go b/internal/fs/task_path_test.go index 767c79881e..e5ac45efe3 100644 --- a/internal/fs/task_path_test.go +++ b/internal/fs/task_path_test.go @@ -1,6 +1,26 @@ package fs -import "testing" +import ( + "testing" + + "github.com/OpenListTeam/OpenList/v4/internal/driver" + "github.com/OpenListTeam/OpenList/v4/internal/model" + "github.com/OpenListTeam/OpenList/v4/internal/task" +) + +type pathTestDriver struct { + driver.Driver + storage model.Storage +} + +func (d *pathTestDriver) GetStorage() *model.Storage { return &d.storage } + +type pathTestFile struct { + model.FileStreamer + name string +} + +func (f *pathTestFile) GetName() string { return f.name } func TestTaskDataGetPaths(t *testing.T) { td := &TaskData{ @@ -31,20 +51,47 @@ func TestTaskDataGetPaths(t *testing.T) { func TestArchiveContentUploadTaskGetPaths(t *testing.T) { t1 := &ArchiveContentUploadTask{ + ObjName: "report.txt", + InPlace: false, DstActualPath: "out", DstStorageMp: "/mount", } if t1.GetSrcPath() != "" { t.Fatal("src should be empty") } - if t1.GetDstPath() != "/mount/out" { + if t1.GetDstPath() != "/mount/out/report.txt" { t.Fatalf("dst = %q", t1.GetDstPath()) } + if !task.MatchTaskPath(t1.GetSrcPath(), t1.GetDstPath(), "/mount/out/report.txt") { + t.Fatal("exact archive upload destination file must match") + } + + inPlace := &ArchiveContentUploadTask{ + ObjName: "expanded-directory", + InPlace: true, + DstActualPath: "out", + DstStorageMp: "/mount", + } + if inPlace.GetDstPath() != "/mount/out" { + t.Fatalf("in-place directory task dst = %q", inPlace.GetDstPath()) + } } -func TestUploadTaskGetPathsNilStorage(t *testing.T) { - t1 := &UploadTask{} - if t1.GetSrcPath() != "" || t1.GetDstPath() != "" { +func TestUploadTaskGetPaths(t *testing.T) { + t1 := &UploadTask{ + storage: &pathTestDriver{storage: model.Storage{MountPath: "/mount"}}, + dstDirActualPath: "uploads", + file: &pathTestFile{name: "photo.jpg"}, + } + if t1.GetDstPath() != "/mount/uploads/photo.jpg" { + t.Fatalf("dst = %q", t1.GetDstPath()) + } + if !task.MatchTaskPath(t1.GetSrcPath(), t1.GetDstPath(), "/mount/uploads/photo.jpg") { + t.Fatal("exact upload destination file must match") + } + + nilTask := &UploadTask{} + if nilTask.GetSrcPath() != "" || nilTask.GetDstPath() != "" { t.Fatal("nil storage upload task paths should be empty") } } diff --git a/server/handles/task.go b/server/handles/task.go index d2070f35e9..00e24c701a 100644 --- a/server/handles/task.go +++ b/server/handles/task.go @@ -37,14 +37,18 @@ type TaskPathReq struct { } type TaskPathResult struct { - Count int `json:"count"` + Matched int `json:"matched"` + Processed int `json:"processed"` } var errEmptyTaskPath = errors.New("path is required") +var errImplicitRootTaskPath = errors.New("root path must be specified explicitly as /") // taskDoneStates are the terminal states of a task. var taskDoneStates = []tache.State{tache.StateCanceled, tache.StateFailed, tache.StateSucceeded} +const taskDeleteWaitTimeout = 30 * time.Second + func getTaskInfo[T task.TaskExtensionInfo](task T) TaskInfo { errMsg := "" if task.GetErr() != nil { @@ -104,10 +108,20 @@ func resolveTaskPathPrefix(user *model.User, raw string) (string, error) { if raw == "" { return "", errEmptyTaskPath } + var resolved string + var err error if user.IsAdmin() { - return utils.FixAndCleanPath(raw), nil + resolved = utils.FixAndCleanPath(raw) + } else { + resolved, err = user.JoinPath(raw) + } + if err != nil { + return "", err } - return user.JoinPath(raw) + if resolved == "/" && raw != "/" { + return "", errImplicitRootTaskPath + } + return resolved, nil } func taskOwnedBy[T task.TaskExtensionInfo](t T, isAdmin bool, uid uint) bool { @@ -171,22 +185,79 @@ func getBatchHandler[T task.TaskExtensionInfo](manager task.Manager[T], callback type pathBatchOp[T task.TaskWithPaths] struct { filter func(t T) bool - apply func(m task.Manager[T], t T) + apply func(m task.Manager[T], t T) bool } -func applyPathBatch[T task.TaskWithPaths](manager task.Manager[T], isAdmin bool, uid uint, prefix string, op pathBatchOp[T]) int { - tasks := manager.GetByCondition(func(t T) bool { - return taskOwnedBy(t, isAdmin, uid) && taskMatchesPathPrefix(t, prefix) && op.filter(t) +func matchingPathTasks[T task.TaskWithPaths](manager task.Manager[T], isAdmin bool, uid uint, prefix string, filter func(T) bool) []T { + return manager.GetByCondition(func(t T) bool { + return taskOwnedBy(t, isAdmin, uid) && taskMatchesPathPrefix(t, prefix) && filter(t) }) - count := 0 +} + +func applyPathBatch[T task.TaskWithPaths](manager task.Manager[T], isAdmin bool, uid uint, prefix string, op pathBatchOp[T]) TaskPathResult { + tasks := matchingPathTasks(manager, isAdmin, uid, prefix, op.filter) + result := TaskPathResult{Matched: len(tasks)} for _, t := range tasks { - op.apply(manager, t) - count++ + if op.apply(manager, t) { + result.Processed++ + } + } + return result +} + +func deletePathBatch[T task.TaskWithPaths](manager task.Manager[T], isAdmin bool, uid uint, prefix string) TaskPathResult { + tasks := matchingPathTasks(manager, isAdmin, uid, prefix, func(T) bool { return true }) + result := TaskPathResult{Matched: len(tasks)} + waitFor := make([]T, 0, len(tasks)) + pending := make(map[string]struct{}, len(tasks)) + + // Cancel every matching task before waiting, so active operations stop in parallel. + for _, selected := range tasks { + current, ok := manager.GetByID(selected.GetID()) + if !ok { + continue + } + state := current.GetState() + if argsContains(state, taskDoneStates...) { + continue + } + manager.Cancel(current.GetID()) + if current.GetStartTime() == nil && argsContains(state, tache.StatePending, tache.StateCanceling, tache.StateWaitingRetry) { + pending[current.GetID()] = struct{}{} + } else { + waitFor = append(waitFor, current) + } + } + + deadline := time.Now().Add(taskDeleteWaitTimeout) + for _, current := range waitFor { + for !argsContains(current.GetState(), taskDoneStates...) && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } } - return count + + for _, selected := range tasks { + current, ok := manager.GetByID(selected.GetID()) + if !ok { + continue + } + state := current.GetState() + // A pending task becomes canceling without running. Its canceled context + // prevents queued execution, so it can be removed immediately. Other tasks + // are removed only after their worker reaches a terminal state. + _, wasPending := pending[current.GetID()] + if !argsContains(state, taskDoneStates...) && !(wasPending && state == tache.StateCanceling) { + continue + } + manager.Remove(current.GetID()) + if _, exists := manager.GetByID(current.GetID()); !exists { + result.Processed++ + } + } + return result } -func getPathBatchHandler[T task.TaskWithPaths](manager task.Manager[T], op pathBatchOp[T]) gin.HandlerFunc { +func getPathBatchHandler[T task.TaskWithPaths](manager task.Manager[T], execute func(task.Manager[T], bool, uint, string) TaskPathResult) gin.HandlerFunc { return func(c *gin.Context) { user, ok := getUser(c) if !ok { @@ -203,44 +274,50 @@ func getPathBatchHandler[T task.TaskWithPaths](manager task.Manager[T], op pathB common.ErrorStrResp(c, err.Error(), 400) return } - count := applyPathBatch(manager, user.IsAdmin(), user.ID, prefix, op) - common.SuccessResp(c, TaskPathResult{Count: count}) + common.SuccessResp(c, execute(manager, user.IsAdmin(), user.ID, prefix)) + } +} + +func pathBatchExecutor[T task.TaskWithPaths](op pathBatchOp[T]) func(task.Manager[T], bool, uint, string) TaskPathResult { + return func(manager task.Manager[T], isAdmin bool, uid uint, prefix string) TaskPathResult { + return applyPathBatch(manager, isAdmin, uid, prefix, op) } } func deleteByPath[T task.TaskWithPaths](manager task.Manager[T]) gin.HandlerFunc { - return getPathBatchHandler(manager, pathBatchOp[T]{ - filter: func(T) bool { return true }, - apply: func(m task.Manager[T], t T) { - // cancel first so running/pending work stops, then remove from manager - if !argsContains(t.GetState(), taskDoneStates...) { - m.Cancel(t.GetID()) - } - m.Remove(t.GetID()) - }, - }) + return getPathBatchHandler(manager, deletePathBatch[T]) } func cancelByPath[T task.TaskWithPaths](manager task.Manager[T]) gin.HandlerFunc { - return getPathBatchHandler(manager, pathBatchOp[T]{ + return getPathBatchHandler(manager, pathBatchExecutor(pathBatchOp[T]{ filter: func(t T) bool { return !argsContains(t.GetState(), taskDoneStates...) }, - apply: func(m task.Manager[T], t T) { - m.Cancel(t.GetID()) + apply: func(m task.Manager[T], selected T) bool { + current, ok := m.GetByID(selected.GetID()) + if !ok || argsContains(current.GetState(), taskDoneStates...) { + return false + } + m.Cancel(current.GetID()) + return true }, - }) + })) } func retryByPath[T task.TaskWithPaths](manager task.Manager[T]) gin.HandlerFunc { - return getPathBatchHandler(manager, pathBatchOp[T]{ + return getPathBatchHandler(manager, pathBatchExecutor(pathBatchOp[T]{ filter: func(t T) bool { return t.GetState() == tache.StateFailed }, - apply: func(m task.Manager[T], t T) { - m.Retry(t.GetID()) + apply: func(m task.Manager[T], selected T) bool { + current, ok := m.GetByID(selected.GetID()) + if !ok || current.GetState() != tache.StateFailed { + return false + } + m.Retry(current.GetID()) + return true }, - }) + })) } func taskRoute[T task.TaskWithPaths](g *gin.RouterGroup, manager task.Manager[T]) { diff --git a/server/handles/task_path_http_test.go b/server/handles/task_path_http_test.go index 18453f2f9b..87d988a8e0 100644 --- a/server/handles/task_path_http_test.go +++ b/server/handles/task_path_http_test.go @@ -56,17 +56,21 @@ func respCode(out map[string]any) int { return -1 } -func respCount(t *testing.T, out map[string]any) int { +func respResult(t *testing.T, out map[string]any) TaskPathResult { t.Helper() data, ok := out["data"].(map[string]any) if !ok { t.Fatalf("no data object in response: %v", out) } - c, ok := data["count"].(float64) + matched, ok := data["matched"].(float64) if !ok { - t.Fatalf("no count in data: %v", data) + t.Fatalf("no matched in data: %v", data) } - return int(c) + processed, ok := data["processed"].(float64) + if !ok { + t.Fatalf("no processed in data: %v", data) + } + return TaskPathResult{Matched: int(matched), Processed: int(processed)} } func TestDeleteByPathHTTPEmptyPathRejected(t *testing.T) { @@ -81,6 +85,20 @@ func TestDeleteByPathHTTPEmptyPathRejected(t *testing.T) { } } +func TestDeleteByPathHTTPImplicitRootRejected(t *testing.T) { + ensureTaskTestConf() + admin := &model.User{ID: 1, Role: model.ADMIN} + m := newPathManager(1) + r := buildPathRouter(m, admin) + + for _, body := range []string{`{"path":"."}`, `{"path":"//"}`, `{"path":"./"}`} { + _, out := postPath(t, r, "/task/copy/delete_by_path", body) + if got := respCode(out); got != 400 { + t.Fatalf("implicit root %s code = %d, want 400 (%v)", body, got, out) + } + } +} + func TestDeleteByPathHTTPNoUserRejected(t *testing.T) { ensureTaskTestConf() m := newPathManager(1) @@ -111,8 +129,8 @@ func TestDeleteByPathHTTPScopedToOwnerAndBasePath(t *testing.T) { if got := respCode(out); got != 200 { t.Fatalf("code = %d (%v)", got, out) } - if got := respCount(t, out); got != 1 { - t.Fatalf("count = %d, want 1 (only own task under base path)", got) + if got := respResult(t, out); got != (TaskPathResult{Matched: 1, Processed: 1}) { + t.Fatalf("result = %+v, want matched=1 processed=1", got) } if _, ok := m.GetByID(mine.GetID()); ok { t.Fatal("owner task under prefix should be removed") @@ -138,8 +156,8 @@ func TestDeleteByPathHTTPAdminNotEscapedByTraversal(t *testing.T) { _, out := postPath(t, r, "/task/copy/delete_by_path", `{"path":"../../etc/secret"}`) code := respCode(out) if code == 200 { - if got := respCount(t, out); got != 0 { - t.Fatalf("SECURITY: traversal deleted %d task(s) outside base path", got) + if got := respResult(t, out); got != (TaskPathResult{}) { + t.Fatalf("SECURITY: traversal processed tasks outside base path: %+v", got) } } if _, ok := m.GetByID(foreign.GetID()); !ok { @@ -159,16 +177,16 @@ func TestCancelAndRetryByPathHTTPStateFiltering(t *testing.T) { r := buildPathRouter(m, admin) _, out := postPath(t, r, "/task/copy/cancel_by_path", `{"path":"/p"}`) - if got := respCount(t, out); got != 1 { - t.Fatalf("cancel count = %d, want 1 (pending only)", got) + if got := respResult(t, out); got != (TaskPathResult{Matched: 1, Processed: 1}) { + t.Fatalf("cancel result = %+v", got) } if done.GetState() != tache.StateSucceeded { t.Fatalf("succeeded task altered by cancel: %v", done.GetState()) } _, out = postPath(t, r, "/task/copy/retry_by_path", `{"path":"/p"}`) - if got := respCount(t, out); got != 1 { - t.Fatalf("retry count = %d, want 1 (failed only)", got) + if got := respResult(t, out); got != (TaskPathResult{Matched: 1, Processed: 1}) { + t.Fatalf("retry result = %+v", got) } if failed.GetState() != tache.StateWaitingRetry { t.Fatalf("failed task not queued for retry: %v", failed.GetState()) diff --git a/server/handles/task_path_test.go b/server/handles/task_path_test.go index 3e8def2b1f..7547997d74 100644 --- a/server/handles/task_path_test.go +++ b/server/handles/task_path_test.go @@ -1,7 +1,10 @@ package handles import ( + "context" + "errors" "fmt" + "sync/atomic" "testing" "time" @@ -70,6 +73,17 @@ func TestResolveTaskPathPrefix(t *testing.T) { if _, err := resolveTaskPathPrefix(admin, " "); err == nil { t.Fatal("expected empty path error") } + for _, ambiguousRoot := range []string{".", "//", "///", "./"} { + if _, err := resolveTaskPathPrefix(admin, ambiguousRoot); !errors.Is(err, errImplicitRootTaskPath) { + t.Fatalf("resolve %q error = %v, want %v", ambiguousRoot, err, errImplicitRootTaskPath) + } + } + if got, err := resolveTaskPathPrefix(admin, "/"); err != nil || got != "/" { + t.Fatalf("explicit root = %q, %v", got, err) + } + if got, err := resolveTaskPathPrefix(user, "."); err != nil || got != "/home/user" { + t.Fatalf("user base path = %q, %v", got, err) + } } func TestApplyPathBatchDeleteCancelRetry(t *testing.T) { @@ -86,17 +100,18 @@ func TestApplyPathBatchDeleteCancelRetry(t *testing.T) { falsePrefix := addPathTask(m, admin, "", "/targetx/x", tache.StatePending) // non-admin can only touch own tasks - count := applyPathBatch(m, false, other.ID, "/target", pathBatchOp[*pathTestTask]{ + result := applyPathBatch(m, false, other.ID, "/target", pathBatchOp[*pathTestTask]{ filter: func(*pathTestTask) bool { return true }, - apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { + apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) bool { if !argsContains(tk.GetState(), taskDoneStates...) { manager.Cancel(tk.GetID()) } manager.Remove(tk.GetID()) + return true }, }) - if count != 1 { - t.Fatalf("non-admin delete count = %d want 1", count) + if result != (TaskPathResult{Matched: 1, Processed: 1}) { + t.Fatalf("non-admin delete result = %+v", result) } if _, ok := m.GetByID(otherUser.GetID()); ok { t.Fatal("other user matched task should be removed") @@ -106,16 +121,17 @@ func TestApplyPathBatchDeleteCancelRetry(t *testing.T) { } // admin cancel by path: only unfinished matching - count = applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ + result = applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ filter: func(tk *pathTestTask) bool { return !argsContains(tk.GetState(), taskDoneStates...) }, - apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { + apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) bool { manager.Cancel(tk.GetID()) + return true }, }) - if count != 1 { - t.Fatalf("cancel count = %d want 1 (pending only; failed is done-state)", count) + if result != (TaskPathResult{Matched: 1, Processed: 1}) { + t.Fatalf("cancel result = %+v", result) } // pending becomes canceling/canceled; failed stays failed; done stays if st := matchPending.GetState(); st != tache.StateCanceling && st != tache.StateCanceled { @@ -129,34 +145,27 @@ func TestApplyPathBatchDeleteCancelRetry(t *testing.T) { } // retry by path: only failed - count = applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ + result = applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ filter: func(tk *pathTestTask) bool { return tk.GetState() == tache.StateFailed }, - apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { + apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) bool { manager.Retry(tk.GetID()) + return true }, }) - if count != 1 { - t.Fatalf("retry count = %d want 1", count) + if result != (TaskPathResult{Matched: 1, Processed: 1}) { + t.Fatalf("retry result = %+v", result) } if matchFailed.GetState() != tache.StateWaitingRetry { t.Fatalf("failed after retry state = %v want waiting_retry", matchFailed.GetState()) } // delete by path: cancel then remove all matching - count = applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ - filter: func(*pathTestTask) bool { return true }, - apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { - if !argsContains(tk.GetState(), taskDoneStates...) { - manager.Cancel(tk.GetID()) - } - manager.Remove(tk.GetID()) - }, - }) + result = deletePathBatch[*pathTestTask](m, true, admin.ID, "/target") // matchPending, matchDstDone, matchFailed still present under /target - if count != 3 { - t.Fatalf("delete count = %d want 3", count) + if result != (TaskPathResult{Matched: 3, Processed: 3}) { + t.Fatalf("delete result = %+v", result) } for _, id := range []string{matchPending.GetID(), matchDstDone.GetID(), matchFailed.GetID()} { if _, ok := m.GetByID(id); ok { @@ -193,18 +202,10 @@ func TestApplyPathBatchLargeScale(t *testing.T) { } start := time.Now() - count := applyPathBatch(m, true, admin.ID, "/bulk", pathBatchOp[*pathTestTask]{ - filter: func(*pathTestTask) bool { return true }, - apply: func(manager task.Manager[*pathTestTask], tk *pathTestTask) { - if !argsContains(tk.GetState(), taskDoneStates...) { - manager.Cancel(tk.GetID()) - } - manager.Remove(tk.GetID()) - }, - }) + result := deletePathBatch[*pathTestTask](m, true, admin.ID, "/bulk") elapsed := time.Since(start) - if count != matchN { - t.Fatalf("count = %d want %d", count, matchN) + if result != (TaskPathResult{Matched: matchN, Processed: matchN}) { + t.Fatalf("result = %+v", result) } remain := len(m.GetAll()) if remain != total-matchN { @@ -216,6 +217,101 @@ func TestApplyPathBatchLargeScale(t *testing.T) { t.Logf("deleted %d/%d tasks in %s", matchN, total, elapsed) } +func TestApplyPathBatchDistinguishesMatchedAndProcessed(t *testing.T) { + ensureTaskTestConf() + admin := &model.User{ID: 1, Role: model.ADMIN} + m := newPathManager(1) + selected := addPathTask(m, admin, "", "/target/race", tache.StateFailed) + + result := applyPathBatch(m, true, admin.ID, "/target", pathBatchOp[*pathTestTask]{ + filter: func(*pathTestTask) bool { return true }, + apply: func(manager task.Manager[*pathTestTask], _ *pathTestTask) bool { + // Simulate another request removing the task after initial matching. + manager.Remove(selected.GetID()) + _, stillExists := manager.GetByID(selected.GetID()) + return stillExists + }, + }) + if result != (TaskPathResult{Matched: 1, Processed: 0}) { + t.Fatalf("result = %+v, want matched=1 processed=0", result) + } +} + +type runningPathTestTask struct { + pathTestTask + started chan struct{} + stopped chan struct{} + cleaned chan struct{} + active atomic.Bool +} + +func (t *runningPathTestTask) Run() error { + t.active.Store(true) + defer t.active.Store(false) + t.SetStartTime(time.Now()) + close(t.started) + <-t.Ctx().Done() + close(t.stopped) + return context.Canceled +} + +func (t *runningPathTestTask) OnFailed() { + // Cleanup deliberately takes time so deletePathBatch must wait for it. + time.Sleep(25 * time.Millisecond) + close(t.cleaned) +} + +func TestDeletePathBatchWaitsForRunningTaskCleanup(t *testing.T) { + ensureTaskTestConf() + admin := &model.User{ID: 1, Role: model.ADMIN} + m := tache.NewManager[*runningPathTestTask](tache.WithWorks(1), tache.WithRunning(true)) + running := &runningPathTestTask{ + pathTestTask: pathTestTask{ + TaskExtension: task.TaskExtension{Creator: admin}, + dst: "/target/running", + }, + started: make(chan struct{}), + stopped: make(chan struct{}), + cleaned: make(chan struct{}), + } + m.Add(running) + + select { + case <-running.started: + case <-time.After(time.Second): + t.Fatal("task did not start") + } + + result := deletePathBatch[*runningPathTestTask](m, true, admin.ID, "/target") + if result != (TaskPathResult{Matched: 1, Processed: 1}) { + t.Fatalf("result = %+v", result) + } + select { + case <-running.stopped: + default: + t.Fatal("task was removed before execution stopped") + } + select { + case <-running.cleaned: + default: + t.Fatal("task was removed before cleanup completed") + } + if _, ok := m.GetByID(running.GetID()); ok { + t.Fatal("cleaned task was not removed") + } + + if running.active.Load() { + t.Fatal("task execution is still active after removal") + } + time.Sleep(50 * time.Millisecond) + if running.active.Load() { + t.Fatal("task continued running in the background after removal") + } + if running.GetState() != tache.StateCanceled { + t.Fatalf("final state = %v, want canceled", running.GetState()) + } +} + func TestTaskOwnedBy(t *testing.T) { admin := &model.User{ID: 1, Role: model.ADMIN} user := &model.User{ID: 2, Role: model.GENERAL}