feat(tasks): cancel queued and suspending tasks (#2270)

Users (and admins) can now cancel tasks that have not started yet.

- queue: StatusQueued/StatusSuspending -> StatusCanceled transitions;
  workers skip canceled tasks at pickup; Queue.CancelTask cancels via
  the in-memory registry
- workflow: POST /workflow/:id/cancel — owner/admin gated; registry
  miss falls back to a persisted status update so restarted queues
  won't resume it; running remote downloads route to CancelDownload
- task detail: cancel action on queued/suspending tasks
- tests: registry cancel, persisted fallback, status/owner gates

Authored By: TDvorak <info@tdvorak.dev>

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

frontend/src/api/api.ts

@@ -919,6 +919,22 @@ export function getTasks(req: ListTaskService): ThunkResponse<TaskListResponse>
   };
 }
 
+export function sendCancelTask(id: string): ThunkResponse<undefined> {
+  return async (dispatch, _getState) => {
+    return await dispatch(
+      send(
+        `/workflow/${id}/cancel`,
+        {
+          method: "POST",
+        },
+        {
+          ...defaultOpts,
+        },
+      ),
+    );
+  };
+}
+
 export function sendRetryTask(id: string): ThunkResponse<undefined> {
   return async (dispatch, _getState) => {
     return await dispatch(

frontend/src/component/Pages/Tasks/TaskDetail.tsx

@@ -12,7 +12,7 @@ import {
 } from "@mui/material";
 import { useState } from "react";
 import { useTranslation } from "react-i18next";
-import { sendRetryTask } from "../../../api/api.ts";
+import { sendCancelTask, sendRetryTask } from "../../../api/api.ts";
 import { TaskResponse, TaskStatus } from "../../../api/workflow.ts";
 import { useAppDispatch } from "../../../redux/hooks.ts";
 import { SecondaryLoadingButton, StyledTableContainerPaper } from "../../Common/StyledComponents.tsx";
@@ -30,6 +30,7 @@ const TaskDetail = ({ task, downloading, onRetried }: TaskDetailProps) => {
   const { t } = useTranslation();
   const dispatch = useAppDispatch();
   const [retrying, setRetrying] = useState(false);
+  const [canceling, setCanceling] = useState(false);
 
   const retry = () => {
     setRetrying(true);
@@ -38,6 +39,16 @@ const TaskDetail = ({ task, downloading, onRetried }: TaskDetailProps) => {
       .catch(() => {})
       .finally(() => setRetrying(false));
   };
+
+  const cancel = () => {
+    setCanceling(true);
+    dispatch(sendCancelTask(task.id))
+      .then(() => onRetried?.())
+      .catch(() => {})
+      .finally(() => setCanceling(false));
+  };
+
+  const cancelable = task.status == TaskStatus.queued || task.status == TaskStatus.suspending;
   return (
     <Stack spacing={2}>
       <Stack spacing={1}>
@@ -60,6 +71,18 @@ const TaskDetail = ({ task, downloading, onRetried }: TaskDetailProps) => {
             })}
           </Alert>
         )}
+        {cancelable && (
+          <SecondaryLoadingButton
+            size="small"
+            variant="outlined"
+            color="error"
+            loading={canceling}
+            onClick={cancel}
+            sx={{ alignSelf: "flex-start" }}
+          >
+            {t("common:cancel")}
+          </SecondaryLoadingButton>
+        )}
         {task.status == TaskStatus.error && (
           <Alert
             severity={"error"}

inventory/task.go

@@ -41,6 +41,8 @@ type TaskClient interface {
 	GetTaskByID(ctx context.Context, taskID int) (*ent.Task, error)
 	// SetCompleteByID sets the task with the given ID to complete.
 	SetCompleteByID(ctx context.Context, taskID int) error
+	// SetStatusByID sets the status of the task with the given ID.
+	SetStatusByID(ctx context.Context, taskID int, status task.Status) error
 	// List returns a list of tasks with the given args.
 	List(ctx context.Context, args *ListTaskArgs) (*ListTaskResult, error)
 	// DeleteByIDs deletes the tasks with the given IDs.
@@ -199,6 +201,13 @@ func (c *taskClient) SetCompleteByID(ctx context.Context, taskID int) error {
 	return err
 }
 
+func (c *taskClient) SetStatusByID(ctx context.Context, taskID int, status task.Status) error {
+	_, err := c.client.Task.UpdateOneID(taskID).
+		SetStatus(status).
+		Save(ctx)
+	return err
+}
+
 func (c *taskClient) List(ctx context.Context, args *ListTaskArgs) (*ListTaskResult, error) {
 	q := c.client.Task.Query()
 	if args.UserID != 0 {

pkg/queue/queue.go

@@ -23,6 +23,10 @@ type (
 		Shutdown()
 		// SubmitTask submits a Task to the queue.
 		QueueTask(ctx context.Context, t Task) error
+		// CancelTask marks a queued or suspending task as canceled. The
+		// scheduler entry is left in place; workers skip canceled tasks at
+		// pickup. Returns false if the task is not in this queue's registry.
+		CancelTask(ctx context.Context, taskID int) bool
 		// BusyWorkers returns the numbers of workers in the running process.
 		BusyWorkers() int
 		// BusyWorkers returns the numbers of success tasks.
@@ -218,6 +222,22 @@ func (q *queue) newContext(t Task) context.Context {
 	return ctx
 }
 
+// CancelTask marks a queued or suspending task as canceled. The scheduler
+// heap entry is not removed; workers skip canceled tasks at pickup.
+func (q *queue) CancelTask(ctx context.Context, taskID int) bool {
+	if q.registry == nil {
+		return false
+	}
+	t, found := q.registry.Get(taskID)
+	if !found {
+		return false
+	}
+	if t.Status() != task.StatusQueued && t.Status() != task.StatusSuspending {
+		return false
+	}
+	return q.transitStatus(ctx, t, task.StatusCanceled) == nil
+}
+
 func (q *queue) work(t Task) {
 	ctx := q.newContext(t)
 	l := logging.FromContext(ctx)
@@ -238,6 +258,11 @@ func (q *queue) work(t Task) {
 		q.schedule()
 	}()
 
+	// Canceled while waiting in the scheduler; nothing to run.
+	if t.Status() == task.StatusCanceled {
+		return
+	}
+
 	err = q.transitStatus(ctx, t, task.StatusProcessing)
 	if err != nil {
 		l.Error("failed to transit task %d to processing: %s", t.ID(), err.Error())

pkg/queue/task.go

@@ -393,6 +393,12 @@ func init() {
 				q.metric.IncFailureTask()
 				return persistTask(ctx, task, newStatus, q)
 			},
+			task.StatusCanceled: func(ctx context.Context, task Task, newStatus task.Status, q *queue) error {
+				if q.registry != nil {
+					q.registry.Delete(task.ID())
+				}
+				return persistTask(ctx, task, newStatus, q)
+			},
 		},
 		task.StatusProcessing: {
 			task.StatusQueued: persistTask,
@@ -468,6 +474,13 @@ func init() {
 				q.metric.IncFailureTask()
 				return persistTask(ctx, task, newStatus, q)
 			},
+			task.StatusCanceled: func(ctx context.Context, task Task, newStatus task.Status, q *queue) error {
+				q.metric.DecSuspendingTask()
+				if q.registry != nil {
+					q.registry.Delete(task.ID())
+				}
+				return persistTask(ctx, task, newStatus, q)
+			},
 		},
 		task.StatusError: {
 			// Manual retry re-queues a failed task with its original args

routers/controllers/workflow.go

@@ -67,6 +67,19 @@ func CancelDownloadTask(c *gin.Context) {
 	c.JSON(200, serializer.Response{})
 }
 
+// CancelTask terminates a queued or suspending task.
+func CancelTask(c *gin.Context) {
+	taskId := hashid.FromContext(c)
+	err := explorer.CancelTask(c, taskId)
+	if err != nil {
+		c.JSON(200, serializer.Err(c, err))
+		c.Abort()
+		return
+	}
+
+	c.JSON(200, serializer.Response{})
+}
+
 // RetryTask re-queues a failed task with its original args.
 func RetryTask(c *gin.Context) {
 	taskId := hashid.FromContext(c)

routers/router.go

@@ -596,6 +596,12 @@ func initMasterRouter(dep dependency.Dep) *gin.Engine {
 				middleware.HashID(hashid.TaskID),
 				controllers.RetryTask,
 			)
+			// Cancel a queued or suspending task
+			wf.POST(":id/cancel",
+				middleware.RequiredScopes(types.ScopeWorkflowWrite),
+				middleware.HashID(hashid.TaskID),
+				controllers.CancelTask,
+			)
 			// Create task to create an archive file
 			wf.POST("archive",
 				middleware.RequiredScopes(types.ScopeWorkflowWrite),

service/explorer/workflows.go

@@ -470,6 +470,55 @@ func RetryTask(c *gin.Context, taskID int) error {
 	return nil
 }
 
+// CancelTask terminates a queued or suspending task; running remote
+// downloads are canceled through their downloader handle (#2270).
+func CancelTask(c *gin.Context, taskID int) error {
+	dep := dependency.FromContext(c)
+	u := inventory.UserFromContext(c)
+	taskClient := dep.TaskClient()
+
+	ctx := context.WithValue(c, inventory.LoadTaskUser{}, 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)
+	}
+
+	switch model.Status {
+	case task.StatusQueued, task.StatusSuspending:
+		q, err := queueForTaskType(c, dep, model.Type)
+		if err != nil {
+			return serializer.NewError(serializer.CodeParamErr, "Task type cannot be canceled", err)
+		}
+		if q.CancelTask(ctx, taskID) {
+			return nil
+		}
+		// Not in the in-memory registry (e.g. after a restart before
+		// resume); persist the cancel so it won't be picked up later.
+		if err := taskClient.SetStatusByID(ctx, taskID, task.StatusCanceled); err != nil {
+			return serializer.NewError(serializer.CodeDBError, "Failed to cancel task", err)
+		}
+		return nil
+	case task.StatusProcessing:
+		// Only remote downloads expose a runtime cancel handle; other
+		// running tasks cannot be interrupted safely.
+		if t, found := dep.TaskRegistry().Get(taskID); found {
+			if dl, ok := t.(*workflows.RemoteDownloadTask); ok {
+				if err := dl.CancelDownload(c); err != nil {
+					return serializer.NewError(serializer.CodeInternalSetting, "Failed to cancel download task", err)
+				}
+				return nil
+			}
+		}
+		return serializer.NewError(serializer.CodeParamErr, "Running task cannot be canceled", nil)
+	default:
+		return serializer.NewError(serializer.CodeParamErr, "Only queued or running tasks can be canceled", nil)
+	}
+}
+
 type (
 	SetDownloadFilesService struct {
 		Files []*downloader.SetFileToDownloadArgs `json:"files" binding:"required"`

service/explorer/workflows_retry_test.go

@@ -24,7 +24,8 @@ import (
 // captureQueue records submitted tasks without running workers.
 type captureQueue struct {
 	queue.Queue
-	submitted []queue.Task
+	submitted  []queue.Task
+	cancelable bool
 }
 
 func (q *captureQueue) QueueTask(ctx context.Context, t queue.Task) error {
@@ -32,7 +33,11 @@ func (q *captureQueue) QueueTask(ctx context.Context, t queue.Task) error {
 	return nil
 }
 
-// retryDepStub exposes only the dependencies RetryTask touches.
+func (q *captureQueue) CancelTask(ctx context.Context, taskID int) bool {
+	return q.cancelable
+}
+
+// retryDepStub exposes only the dependencies RetryTask/CancelTask touch.
 type retryDepStub struct {
 	dependency.Dep
 	taskClient inventory.TaskClient
@@ -40,6 +45,7 @@ type retryDepStub struct {
 	downloadQ  queue.Queue
 	recycleQ   queue.Queue
 	mediaMetaQ queue.Queue
+	registry   queue.TaskRegistry
 }
 
 func (d *retryDepStub) TaskClient() inventory.TaskClient                { return d.taskClient }
@@ -47,6 +53,12 @@ func (d *retryDepStub) IoIntenseQueue(context.Context) queue.Queue      { return
 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 }
+func (d *retryDepStub) TaskRegistry() queue.TaskRegistry {
+	if d.registry == nil {
+		d.registry = queue.NewTaskRegistry()
+	}
+	return d.registry
+}
 
 // TestRetryTask verifies failed-task retry: owner/admin gating, status gate,
 // and queue routing by task type (upstream #2823).
@@ -128,3 +140,88 @@ func TestRetryTask(t *testing.T) {
 		require.Error(t, RetryTask(newCtx(owner), m.ID))
 	})
 }
+
+// TestCancelTask verifies queued-task cancellation falls back to a persisted
+// status update when the task is not in the queue's in-memory registry
+// (upstream #2270).
+func TestCancelTask(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)
+	owner := client.User.Create().SetEmail("cancel-owner@example.com").SetNick("u").SetStatus("active").SetGroup(group).SaveX(ctx)
+	owner = client.User.Query().WithGroup().Where(entuser.ID(owner.ID)).OnlyX(ctx)
+	other := client.User.Create().SetEmail("cancel-other@example.com").SetNick("u").SetStatus("active").SetGroup(group).SaveX(ctx)
+	other = client.User.Query().WithGroup().Where(entuser.ID(other.ID)).OnlyX(ctx)
+
+	ioQ := &captureQueue{}
+	dep := &retryDepStub{
+		taskClient: inventory.NewTaskClient(client, conf.SQLiteDB, nil),
+		ioQueue:    ioQ,
+		downloadQ:  &captureQueue{},
+		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("queued task canceled via registry", func(t *testing.T) {
+		ioQ.cancelable = true
+		m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusQueued)
+		require.NoError(t, CancelTask(newCtx(owner), m.ID))
+	})
+
+	t.Run("queued task falls back to persisted cancel", func(t *testing.T) {
+		ioQ.cancelable = false
+		m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusQueued)
+		require.NoError(t, CancelTask(newCtx(owner), m.ID))
+		require.Equal(t, enttask.StatusCanceled, client.Task.GetX(ctx, m.ID).Status)
+	})
+
+	t.Run("suspending task cancelable", func(t *testing.T) {
+		ioQ.cancelable = false
+		m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusSuspending)
+		require.NoError(t, CancelTask(newCtx(owner), m.ID))
+		require.Equal(t, enttask.StatusCanceled, client.Task.GetX(ctx, m.ID).Status)
+	})
+
+	t.Run("completed task rejected", func(t *testing.T) {
+		m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusCompleted)
+		require.Error(t, CancelTask(newCtx(owner), m.ID))
+	})
+
+	t.Run("processing non-download task rejected", func(t *testing.T) {
+		m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusProcessing)
+		require.Error(t, CancelTask(newCtx(owner), m.ID))
+	})
+
+	t.Run("other user's task rejected", func(t *testing.T) {
+		m := newTask(owner.ID, queue.ExtractArchiveTaskType, enttask.StatusQueued)
+		require.Error(t, CancelTask(newCtx(other), m.ID))
+	})
+
+	t.Run("slave task type not cancelable", func(t *testing.T) {
+		m := newTask(owner.ID, queue.SlaveUploadTaskType, enttask.StatusQueued)
+		require.Error(t, CancelTask(newCtx(owner), m.ID))
+	})
+}