diff --git a/frontend/src/api/api.ts b/frontend/src/api/api.ts index cb145a04..1d920060 100644 --- a/frontend/src/api/api.ts +++ b/frontend/src/api/api.ts @@ -919,6 +919,22 @@ export function getTasks(req: ListTaskService): ThunkResponse }; } +export function sendRetryTask(id: string): ThunkResponse { + return async (dispatch, _getState) => { + return await dispatch( + send( + `/workflow/${id}/retry`, + { + method: "POST", + }, + { + ...defaultOpts, + }, + ), + ); + }; +} + export function getTasksPhaseProgress(id: string): ThunkResponse { return async (dispatch, _getState) => { return await dispatch( diff --git a/frontend/src/component/Pages/Tasks/DownloadList.tsx b/frontend/src/component/Pages/Tasks/DownloadList.tsx index a5ba2b8f..c292181a 100644 --- a/frontend/src/component/Pages/Tasks/DownloadList.tsx +++ b/frontend/src/component/Pages/Tasks/DownloadList.tsx @@ -140,12 +140,13 @@ const DownloadList = () => { )} {downloadingTasks == undefined && } - {downloadingTasks && downloadingTasks.map((task) => )} + {downloadingTasks && + downloadingTasks.map((task) => )} {t("download.finished")} {tasks.map((task) => ( - + ))} {nextPageToken != undefined && ( diff --git a/frontend/src/component/Pages/Tasks/TaskCard.tsx b/frontend/src/component/Pages/Tasks/TaskCard.tsx index cfd468fc..31c9c57d 100644 --- a/frontend/src/component/Pages/Tasks/TaskCard.tsx +++ b/frontend/src/component/Pages/Tasks/TaskCard.tsx @@ -96,6 +96,7 @@ export interface TaskCardProps { showProgress?: boolean; task?: TaskResponse; onLoad?: () => void; + onRetried?: () => void; } const taskIconsMap: { @@ -107,7 +108,7 @@ const taskIconsMap: { [TaskType.import]: ArrowImport, }; -const TaskCard = ({ loading, showProgress, onLoad, task }: TaskCardProps) => { +const TaskCard = ({ loading, showProgress, onLoad, onRetried, task }: TaskCardProps) => { const { t } = useTranslation(); const theme = useTheme(); const isMobile = useMediaQuery(theme.breakpoints.down("md")); @@ -214,7 +215,7 @@ const TaskCard = ({ loading, showProgress, onLoad, task }: TaskCardProps) => { - {task && } + {task && } ); }; diff --git a/frontend/src/component/Pages/Tasks/TaskDetail.tsx b/frontend/src/component/Pages/Tasks/TaskDetail.tsx index d3fd7c20..9979ae15 100644 --- a/frontend/src/component/Pages/Tasks/TaskDetail.tsx +++ b/frontend/src/component/Pages/Tasks/TaskDetail.tsx @@ -10,9 +10,12 @@ import { TableRow, Typography, } from "@mui/material"; +import { useState } from "react"; import { useTranslation } from "react-i18next"; +import { sendRetryTask } from "../../../api/api.ts"; import { TaskResponse, TaskStatus } from "../../../api/workflow.ts"; -import { StyledTableContainerPaper } from "../../Common/StyledComponents.tsx"; +import { useAppDispatch } from "../../../redux/hooks.ts"; +import { SecondaryLoadingButton, StyledTableContainerPaper } from "../../Common/StyledComponents.tsx"; import DownloadFileList from "./DownloadFileList.tsx"; import TaskProgress from "./TaskProgress.tsx"; import TaskProps from "./TaskProps.tsx"; @@ -20,10 +23,21 @@ import TaskProps from "./TaskProps.tsx"; export interface TaskDetailProps { task: TaskResponse; downloading?: boolean; + onRetried?: () => void; } -const TaskDetail = ({ task, downloading }: TaskDetailProps) => { +const TaskDetail = ({ task, downloading, onRetried }: TaskDetailProps) => { const { t } = useTranslation(); + const dispatch = useAppDispatch(); + const [retrying, setRetrying] = useState(false); + + const retry = () => { + setRetrying(true); + dispatch(sendRetryTask(task.id)) + .then(() => onRetried?.()) + .catch(() => {}) + .finally(() => setRetrying(false)); + }; return ( @@ -46,7 +60,24 @@ const TaskDetail = ({ task, downloading }: TaskDetailProps) => { })} )} - {task.status == TaskStatus.error && {task.error}} + {task.status == TaskStatus.error && ( + + {t("uploader.retry")} + + } + > + {task.error} + + )} { title={t("application:navbar.taskQueue")} /> {tasks.map((task) => ( - + ))} {nextPageToken != undefined && ( diff --git a/pkg/queue/task.go b/pkg/queue/task.go index 0a2d9fc9..eeb57bbb 100644 --- a/pkg/queue/task.go +++ b/pkg/queue/task.go @@ -469,6 +469,11 @@ func init() { return persistTask(ctx, task, newStatus, q) }, }, + task.StatusError: { + // Manual retry re-queues a failed task with its original args + // (#2823). RetryCount history is preserved in PublicState. + task.StatusQueued: persistTask, + }, } } diff --git a/routers/controllers/workflow.go b/routers/controllers/workflow.go index f2a8a3a2..95ddda54 100644 --- a/routers/controllers/workflow.go +++ b/routers/controllers/workflow.go @@ -66,3 +66,16 @@ func CancelDownloadTask(c *gin.Context) { c.JSON(200, serializer.Response{}) } + +// RetryTask re-queues a failed task with its original args. +func RetryTask(c *gin.Context) { + taskId := hashid.FromContext(c) + err := explorer.RetryTask(c, taskId) + if err != nil { + c.JSON(200, serializer.Err(c, err)) + c.Abort() + return + } + + c.JSON(200, serializer.Response{}) +} diff --git a/routers/router.go b/routers/router.go index 0b124e06..f2141512 100644 --- a/routers/router.go +++ b/routers/router.go @@ -590,6 +590,12 @@ func initMasterRouter(dep dependency.Dep) *gin.Engine { middleware.HashID(hashid.TaskID), controllers.GetTaskPhaseProgress, ) + // Retry a failed task with its original args + wf.POST(":id/retry", + middleware.RequiredScopes(types.ScopeWorkflowWrite), + middleware.HashID(hashid.TaskID), + controllers.RetryTask, + ) // Create task to create an archive file wf.POST("archive", middleware.RequiredScopes(types.ScopeWorkflowWrite), diff --git a/service/explorer/workflows.go b/service/explorer/workflows.go index 8bef8c95..680e572a 100644 --- a/service/explorer/workflows.go +++ b/service/explorer/workflows.go @@ -1,7 +1,9 @@ package explorer import ( + "context" "encoding/gob" + "fmt" "github.com/cloudreve/Cloudreve/v4/pkg/hashid" "time" @@ -413,6 +415,61 @@ func CancelDownloadTask(c *gin.Context, taskID int) error { return nil } +// queueForTaskType maps a persisted task type to the dependency queue that +// owns it. Slave-side task types are not retryable from the master UI. +func queueForTaskType(c *gin.Context, dep dependency.Dep, taskType string) (queue.Queue, error) { + switch taskType { + case queue.CreateArchiveTaskType, queue.ExtractArchiveTaskType, queue.RelocateTaskType, queue.ImportTaskType: + return dep.IoIntenseQueue(c), nil + case queue.RemoteDownloadTaskType: + return dep.RemoteDownloadQueue(c), nil + case queue.MediaMetaTaskType, queue.FullTextIndexTaskType, queue.FullTextDeleteTaskType, + queue.FullTextRebuildTaskType, queue.FullTextCopyTaskType, queue.FullTextChangeOwnerTaskType: + return dep.MediaMetaQueue(c), nil + case queue.EntityRecycleRoutineTaskType, queue.ExplicitEntityRecycleTaskType, queue.UploadSentinelCheckTaskType: + return dep.EntityRecycleQueue(c), nil + } + return nil, fmt.Errorf("task type %q is not retryable", taskType) +} + +// RetryTask re-queues a failed task with its original args (#2823). +func RetryTask(c *gin.Context, taskID int) error { + dep := dependency.FromContext(c) + u := inventory.UserFromContext(c) + taskClient := dep.TaskClient() + + ctx := context.WithValue(c, inventory.LoadTaskUser{}, true) + ctx = context.WithValue(ctx, inventory.LoadUserGroup{}, true) + model, err := taskClient.GetTaskByID(ctx, taskID) + if err != nil { + return serializer.NewError(serializer.CodeNotFound, "Task not found", err) + } + + if model.UserTasks != u.ID && !u.Edges.Group.Permissions.Enabled(int(types.GroupPermissionIsAdmin)) { + return serializer.NewError(serializer.CodeNotFound, "Task not found", nil) + } + + if model.Status != task.StatusError { + return serializer.NewError(serializer.CodeParamErr, "Only failed tasks can be retried", nil) + } + + resumed, err := queue.NewTaskFromModel(model) + if err != nil { + return serializer.NewError(serializer.CodeInternalSetting, "Failed to rebuild task", err) + } + + q, err := queueForTaskType(c, dep, model.Type) + if err != nil { + return serializer.NewError(serializer.CodeParamErr, "Task type is not retryable", err) + } + + if err := q.QueueTask(c, resumed); err != nil { + return serializer.NewError(serializer.CodeCreateTaskError, "Failed to queue task", err) + } + + return nil +} + type ( SetDownloadFilesService struct { Files []*downloader.SetFileToDownloadArgs `json:"files" binding:"required"` diff --git a/service/explorer/workflows_retry_test.go b/service/explorer/workflows_retry_test.go new file mode 100644 index 00000000..00a6b459 --- /dev/null +++ b/service/explorer/workflows_retry_test.go @@ -0,0 +1,130 @@ +package explorer + +import ( + "context" + "net/http/httptest" + "testing" + + "github.com/cloudreve/Cloudreve/v4/application/dependency" + "github.com/cloudreve/Cloudreve/v4/ent" + "github.com/cloudreve/Cloudreve/v4/ent/enttest" + enttask "github.com/cloudreve/Cloudreve/v4/ent/task" + entuser "github.com/cloudreve/Cloudreve/v4/ent/user" + "github.com/cloudreve/Cloudreve/v4/inventory" + "github.com/cloudreve/Cloudreve/v4/inventory/types" + "github.com/cloudreve/Cloudreve/v4/pkg/boolset" + "github.com/cloudreve/Cloudreve/v4/pkg/conf" + "github.com/cloudreve/Cloudreve/v4/pkg/queue" + "github.com/cloudreve/Cloudreve/v4/pkg/util" + "github.com/gin-gonic/gin" + "github.com/gofrs/uuid" + "github.com/stretchr/testify/require" +) + +// captureQueue records submitted tasks without running workers. +type captureQueue struct { + queue.Queue + submitted []queue.Task +} + +func (q *captureQueue) QueueTask(ctx context.Context, t queue.Task) error { + q.submitted = append(q.submitted, t) + return nil +} + +// retryDepStub exposes only the dependencies RetryTask touches. +type retryDepStub struct { + dependency.Dep + taskClient inventory.TaskClient + ioQueue queue.Queue + downloadQ queue.Queue + recycleQ queue.Queue + mediaMetaQ queue.Queue +} + +func (d *retryDepStub) TaskClient() inventory.TaskClient { return d.taskClient } +func (d *retryDepStub) IoIntenseQueue(context.Context) queue.Queue { return d.ioQueue } +func (d *retryDepStub) RemoteDownloadQueue(context.Context) queue.Queue { return d.downloadQ } +func (d *retryDepStub) EntityRecycleQueue(context.Context) queue.Queue { return d.recycleQ } +func (d *retryDepStub) MediaMetaQueue(context.Context) queue.Queue { return d.mediaMetaQ } + +// TestRetryTask verifies failed-task retry: owner/admin gating, status gate, +// and queue routing by task type (upstream #2823). +func TestRetryTask(t *testing.T) { + gin.SetMode(gin.TestMode) + client := enttest.Open(t, "sqlite3", "file:"+t.Name()+"?mode=memory&cache=shared") + t.Cleanup(func() { require.NoError(t, client.Close()) }) + ctx := context.Background() + + group := client.Group.Create().SetName("g").SetPermissions(&boolset.BooleanSet{}).SaveX(ctx) + newUser := func(email string) *ent.User { + u := client.User.Create().SetEmail(email).SetNick("u").SetStatus("active").SetGroup(group).SaveX(ctx) + return client.User.Query().WithGroup().Where(entuser.ID(u.ID)).OnlyX(ctx) + } + owner := newUser("owner@example.com") + other := newUser("other@example.com") + + ioQ := &captureQueue{} + downloadQ := &captureQueue{} + dep := &retryDepStub{ + taskClient: inventory.NewTaskClient(client, conf.SQLiteDB, nil), + ioQueue: ioQ, + downloadQ: downloadQ, + recycleQ: &captureQueue{}, + mediaMetaQ: &captureQueue{}, + } + + newCtx := func(u *ent.User) *gin.Context { + engine := gin.New() + engine.ContextWithFallback = true + c := gin.CreateTestContextOnly(httptest.NewRecorder(), engine) + c.Request = httptest.NewRequest("POST", "/", nil) + util.WithValue(c, dependency.DepCtx{}, dep) + util.WithValue(c, inventory.UserCtx{}, u) + return c + } + + newTask := func(ownerID int, taskType string, status enttask.Status) *ent.Task { + return client.Task.Create(). + SetType(taskType). + SetStatus(status). + SetPublicState(&types.TaskPublicState{}). + SetCorrelationID(uuid.Must(uuid.NewV4())). + SetUserTasks(ownerID). + SaveX(ctx) + } + + t.Run("failed extract task re-queues to io-intense queue", func(t *testing.T) { + ioQ.submitted = nil + m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusError) + require.NoError(t, RetryTask(newCtx(owner), m.ID)) + require.Len(t, ioQ.submitted, 1) + require.Equal(t, m.ID, ioQ.submitted[0].ID()) + }) + + t.Run("failed download task re-queues to download queue", func(t *testing.T) { + downloadQ.submitted = nil + m := newTask(owner.ID, queue.RemoteDownloadTaskType, enttask.StatusError) + require.NoError(t, RetryTask(newCtx(owner), m.ID)) + require.Len(t, downloadQ.submitted, 1) + }) + + t.Run("non-error task rejected", func(t *testing.T) { + ioQ.submitted = nil + m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusCompleted) + require.Error(t, RetryTask(newCtx(owner), m.ID)) + require.Empty(t, ioQ.submitted) + }) + + t.Run("other user's task rejected", func(t *testing.T) { + ioQ.submitted = nil + m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusError) + require.Error(t, RetryTask(newCtx(other), m.ID)) + require.Empty(t, ioQ.submitted) + }) + + t.Run("slave task type not retryable", func(t *testing.T) { + m := newTask(owner.ID, queue.SlaveUploadTaskType, enttask.StatusError) + require.Error(t, RetryTask(newCtx(owner), m.ID)) + }) +}