From a2e46faab9f4b0f013f52e7b1cce59081685d133 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=80=97=E5=AD=90?= Date: Mon, 20 Jul 2026 23:16:49 +0800 Subject: [PATCH] =?UTF-8?q?feat(task):=20=E4=BB=BB=E5=8A=A1=E9=98=9F?= =?UTF-8?q?=E5=88=97=E6=94=AF=E6=8C=81=E5=8F=96=E6=B6=88=E8=BF=90=E8=A1=8C?= =?UTF-8?q?=E4=B8=AD=E4=B8=8E=E7=AD=89=E5=BE=85=E4=B8=AD=E7=9A=84=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 任务执行接入 context 并放入独立进程组,取消时整组杀死; 新增可选 CancelShell 清理钩子,远程下载、压缩、备份任务 取消后自动清理残留文件,备份改用独享临时目录避免误伤并发备份; 防重改为按语言无关的任务标识 Key 匹配 Co-Authored-By: Claude Fable 5 --- internal/apps/prometheus/app.go | 2 + internal/biz/app.go | 3 + internal/biz/environment.go | 1 + internal/biz/task.go | 30 +++++--- internal/data/task.go | 57 ++++++++++++--- internal/data/task_test.go | 55 ++++++++++++++ internal/migration/v1.go | 9 +++ internal/route/task.go | 1 + internal/service/backup.go | 24 +++++-- internal/service/environment_php.go | 2 + internal/service/file.go | 5 ++ internal/service/task.go | 15 ++++ internal/taskqueue/runner.go | 82 ++++++++++++++++++--- internal/taskqueue/runner_test.go | 108 ++++++++++++++++++++++++++++ mocks/biz/TaskRepo.go | 46 ++++++++++++ pkg/shell/exec.go | 14 +++- pkg/types/task.go | 1 + web/src/api/panel/task/index.ts | 2 + web/src/views/task/TaskView.vue | 38 ++++++++-- 19 files changed, 455 insertions(+), 40 deletions(-) create mode 100644 internal/data/task_test.go create mode 100644 internal/taskqueue/runner_test.go diff --git a/internal/apps/prometheus/app.go b/internal/apps/prometheus/app.go index 6ddc90b3..408c1def 100644 --- a/internal/apps/prometheus/app.go +++ b/internal/apps/prometheus/app.go @@ -263,6 +263,7 @@ func (s *App) InstallExporter(w http.ResponseWriter, r *http.Request) { cmd := fmt.Sprintf(`curl -sSLm 10 --retry 3 'https://%s/prometheus/exporters/exporter.sh' | bash -s -- 'install' '%s'`, s.conf.App.DownloadEndpoint, url.PathEscape(req.Slug)) task := new(biz.Task) + task.Key = "prometheus:exporter:" + req.Slug task.Name = s.t.Get("Install Prometheus exporter %s", req.Slug) task.Status = biz.TaskStatusWaiting task.Shell = cmd @@ -290,6 +291,7 @@ func (s *App) UninstallExporter(w http.ResponseWriter, r *http.Request) { cmd := fmt.Sprintf(`curl -sSLm 10 --retry 3 'https://%s/prometheus/exporters/exporter.sh' | bash -s -- 'uninstall' '%s'`, s.conf.App.DownloadEndpoint, url.PathEscape(req.Slug)) task := new(biz.Task) + task.Key = "prometheus:exporter:" + req.Slug task.Name = s.t.Get("Uninstall Prometheus exporter %s", req.Slug) task.Status = biz.TaskStatusWaiting task.Shell = cmd diff --git a/internal/biz/app.go b/internal/biz/app.go index dd743079..d4299f83 100644 --- a/internal/biz/app.go +++ b/internal/biz/app.go @@ -213,6 +213,7 @@ func (uc *AppUsecase) Install(channel, slug string) error { } task := new(Task) + task.Key = "app:" + slug task.Name = uc.t.Get("Install app %s", item.Name) task.Status = TaskStatusWaiting task.Shell = script @@ -253,6 +254,7 @@ func (uc *AppUsecase) UnInstall(slug string) error { } task := new(Task) + task.Key = "app:" + slug task.Name = uc.t.Get("Uninstall app %s", item.Name) task.Status = TaskStatusWaiting task.Shell = script @@ -295,6 +297,7 @@ func (uc *AppUsecase) Update(slug string) error { } task := new(Task) + task.Key = "app:" + slug task.Name = uc.t.Get("Update app %s", item.Name) task.Status = TaskStatusWaiting task.Shell = script diff --git a/internal/biz/environment.go b/internal/biz/environment.go index b8cb8dce..b88d2506 100644 --- a/internal/biz/environment.go +++ b/internal/biz/environment.go @@ -157,6 +157,7 @@ func (uc *EnvironmentUsecase) do(typ, slug, action string) error { } task := new(Task) + task.Key = fmt.Sprintf("environment:%s:%s", typ, slug) task.Name = name task.Status = TaskStatusWaiting task.Shell = cmd diff --git a/internal/biz/task.go b/internal/biz/task.go index f24c3b6f..7d2b9125 100644 --- a/internal/biz/task.go +++ b/internal/biz/task.go @@ -5,20 +5,23 @@ import "time" type TaskStatus string const ( - TaskStatusWaiting TaskStatus = "waiting" - TaskStatusRunning TaskStatus = "running" - TaskStatusSuccess TaskStatus = "finished" - TaskStatusFailed TaskStatus = "failed" + TaskStatusWaiting TaskStatus = "waiting" + TaskStatusRunning TaskStatus = "running" + TaskStatusSuccess TaskStatus = "finished" + TaskStatusFailed TaskStatus = "failed" + TaskStatusCanceled TaskStatus = "canceled" ) type Task struct { - ID uint `gorm:"primaryKey" json:"id"` - Name string `gorm:"not null;default:'';index" json:"name"` - Status TaskStatus `gorm:"not null;default:'waiting'" json:"status"` - Shell string `gorm:"not null;default:''" json:"-"` - Log string `gorm:"not null;default:''" json:"log"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID uint `gorm:"primaryKey" json:"id"` + Key string `gorm:"not null;default:'';index" json:"-"` // 任务标识（域:动作:标的），用于防重 + Name string `gorm:"not null;default:'';index" json:"name"` + Status TaskStatus `gorm:"not null;default:'waiting'" json:"status"` + Shell string `gorm:"not null;default:''" json:"-"` + CancelShell string `gorm:"not null;default:''" json:"-"` // 运行中被取消后执行的清理命令（可选） + Log string `gorm:"not null;default:''" json:"log"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } type TaskRepo interface { @@ -26,6 +29,7 @@ type TaskRepo interface { List(page, limit uint) ([]*Task, int64, error) Get(id uint) (*Task, error) Delete(id uint) error + Cancel(id uint) error UpdateStatus(id uint, status TaskStatus) error UpdateLog(id uint, log string) error Push(task *Task) error @@ -55,6 +59,10 @@ func (uc *TaskUsecase) Delete(id uint) error { return uc.repo.Delete(id) } +func (uc *TaskUsecase) Cancel(id uint) error { + return uc.repo.Cancel(id) +} + func (uc *TaskUsecase) UpdateStatus(id uint, status TaskStatus) error { return uc.repo.UpdateStatus(id, status) } diff --git a/internal/data/task.go b/internal/data/task.go index a1bef31a..5456027c 100644 --- a/internal/data/task.go +++ b/internal/data/task.go @@ -3,6 +3,7 @@ package data import ( "errors" "log/slog" + "os" "github.com/leonelquinteros/gotext" "github.com/samber/do/v2" @@ -48,7 +49,45 @@ func (r *taskRepo) Get(id uint) (*biz.Task, error) { } func (r *taskRepo) Delete(id uint) error { - return r.db.Model(&biz.Task{}).Where("id = ?", id).Delete(&biz.Task{}).Error + task, err := r.Get(id) + if err != nil { + return err + } + if task.Status == biz.TaskStatusWaiting || task.Status == biz.TaskStatusRunning { + return errors.New(r.t.Get("please cancel the task first")) + } + + // 清理任务日志文件 + if task.Log != "" { + _ = os.Remove(task.Log) + } + + return r.db.Where("id = ?", id).Delete(&biz.Task{}).Error +} + +func (r *taskRepo) Cancel(id uint) error { + // 等待中的任务直接原子标记取消，避免与运行器取任务竞争 + result := r.db.Model(&biz.Task{}).Where("id = ? AND status = ?", id, biz.TaskStatusWaiting).Update("status", biz.TaskStatusCanceled) + if result.Error != nil { + return result.Error + } + if result.RowsAffected > 0 { + return nil + } + + task, err := r.Get(id) + if err != nil { + return err + } + if task.Status != biz.TaskStatusRunning { + return errors.New(r.t.Get("task is not waiting or running")) + } + + // 运行中的任务交由运行器杀死进程组 + if !r.runner.Cancel(id) { + return errors.New(r.t.Get("task has already finished")) + } + return nil } func (r *taskRepo) UpdateStatus(id uint, status biz.TaskStatus) error { @@ -60,13 +99,15 @@ func (r *taskRepo) UpdateLog(id uint, log string) error { } func (r *taskRepo) Push(task *biz.Task) error { - // 防止有人喜欢酒吧点炒饭 - var count int64 - if err := r.db.Model(&biz.Task{}).Where("shell = ? and (status = ? or status = ?)", task.Shell, biz.TaskStatusWaiting, biz.TaskStatusRunning).Count(&count).Error; err != nil { - return err - } - if count > 0 { - return errors.New(r.t.Get("duplicate submission, please wait for the previous task to end")) + // 防止有人喜欢酒吧点炒饭，按语言无关的任务标识去重 + if task.Key != "" { + var count int64 + if err := r.db.Model(&biz.Task{}).Where("`key` = ? and (status = ? or status = ?)", task.Key, biz.TaskStatusWaiting, biz.TaskStatusRunning).Count(&count).Error; err != nil { + return err + } + if count > 0 { + return errors.New(r.t.Get("duplicate submission, please wait for the previous task to end")) + } } if err := r.db.Create(task).Error; err != nil { diff --git a/internal/data/task_test.go b/internal/data/task_test.go new file mode 100644 index 00000000..103e8a78 --- /dev/null +++ b/internal/data/task_test.go @@ -0,0 +1,55 @@ +package data + +import ( + "context" + "testing" + + "github.com/leonelquinteros/gotext" + "github.com/libtnb/sqlite" + "gorm.io/gorm" + + "github.com/acepanel/panel/v3/internal/biz" +) + +type stubRunner struct{} + +func (stubRunner) Run(context.Context) {} +func (stubRunner) Notify() {} +func (stubRunner) Cancel(uint) bool { return false } + +func newTaskRepoForTest(t *testing.T) *taskRepo { + t.Helper() + db, err := gorm.Open(sqlite.Open("file::memory:"), &gorm.Config{SkipDefaultTransaction: true}) + if err != nil { + t.Fatal(err) + } + if err = db.AutoMigrate(&biz.Task{}); err != nil { + t.Fatal(err) + } + return &taskRepo{t: gotext.NewLocale("", "en"), db: db, runner: stubRunner{}} +} + +func TestTaskPushDedup(t *testing.T) { + repo := newTaskRepoForTest(t) + + // shell 含随机片段也按 key 去重 + if err := repo.Push(&biz.Task{Key: "backup:website:a", Name: "备份", Status: biz.TaskStatusWaiting, Shell: "echo 1"}); err != nil { + t.Fatalf("first Push: %v", err) + } + if err := repo.Push(&biz.Task{Key: "backup:website:a", Name: "Backup", Status: biz.TaskStatusWaiting, Shell: "echo 2"}); err == nil { + t.Fatal("duplicate key should be rejected") + } + + // 不同 key 放行 + if err := repo.Push(&biz.Task{Key: "backup:website:b", Name: "Backup", Status: biz.TaskStatusWaiting, Shell: "echo 3"}); err != nil { + t.Fatalf("different key should pass: %v", err) + } + + // 终态后同 key 放行 + if err := repo.db.Model(&biz.Task{}).Where("`key` = ?", "backup:website:a").Update("status", biz.TaskStatusCanceled).Error; err != nil { + t.Fatal(err) + } + if err := repo.Push(&biz.Task{Key: "backup:website:a", Name: "Backup", Status: biz.TaskStatusWaiting, Shell: "echo 4"}); err != nil { + t.Fatalf("Push after terminal status should pass: %v", err) + } +} diff --git a/internal/migration/v1.go b/internal/migration/v1.go index 303b0ffc..e4b5dc12 100644 --- a/internal/migration/v1.go +++ b/internal/migration/v1.go @@ -140,4 +140,13 @@ func init() { return tx.Migrator().DropTable(&biz.FileShare{}) }, }) + Migrations = append(Migrations, &gormigrate.Migration{ + ID: "20260720-update-task-fields", + Migrate: func(tx *gorm.DB) error { + return tx.AutoMigrate(&biz.Task{}) + }, + Rollback: func(tx *gorm.DB) error { + return nil + }, + }) } diff --git a/internal/route/task.go b/internal/route/task.go index ae068abb..5716a8c7 100644 --- a/internal/route/task.go +++ b/internal/route/task.go @@ -19,5 +19,6 @@ func TaskRoutes(i do.Injector) (Endpoints, error) { {Method: http.MethodGet, Path: "/api/task", Handler: svc.List, Summary: "获取任务列表", Tags: []string{"任务"}, Request: request.Paginate{}, Response: service.Envelope[service.Page[*biz.Task]]{}}, {Method: http.MethodGet, Path: "/api/task/{id}", Handler: svc.Get, Summary: "获取任务详情", Tags: []string{"任务"}, Request: request.ID{}, Response: service.Envelope[biz.Task]{}}, {Method: http.MethodDelete, Path: "/api/task/{id}", Handler: svc.Delete, Summary: "删除任务", Tags: []string{"任务"}, Request: request.ID{}}, + {Method: http.MethodPost, Path: "/api/task/{id}/cancel", Handler: svc.Cancel, Summary: "取消任务", Tags: []string{"任务"}, Request: request.ID{}}, }, nil } diff --git a/internal/service/backup.go b/internal/service/backup.go index de3da997..c1c1db80 100644 --- a/internal/service/backup.go +++ b/internal/service/backup.go @@ -10,6 +10,7 @@ import ( "github.com/leonelquinteros/gotext" "github.com/libtnb/chix/v2" + "github.com/libtnb/utils/str" "github.com/samber/do/v2" "github.com/acepanel/panel/v3/internal/biz" @@ -61,17 +62,28 @@ func (s *BackupService) Create(w http.ResponseWriter, r *http.Request) { // 备份可能耗时较长（大库），提交到后台任务队列异步执行 pathEnv := "export PATH=/bin:/sbin:/usr/bin:/usr/sbin:/usr/local/bin:/usr/local/sbin:$PATH\n" - var cmd string + var backupCmd string if req.Type == "website" { - cmd = fmt.Sprintf("%sacepanel backup website -n '%s' -s '%d'", pathEnv, req.Target, req.Storage) + backupCmd = fmt.Sprintf("acepanel backup website -n '%s' -s '%d'", req.Target, req.Storage) } else { - cmd = fmt.Sprintf("%sacepanel backup database -t '%s' -n '%s' -s '%d'", pathEnv, req.Type, req.Target, req.Storage) + backupCmd = fmt.Sprintf("acepanel backup database -t '%s' -n '%s' -s '%d'", req.Type, req.Target, req.Storage) } + // 备份进程被杀后无法自清临时目录，包一层独享 TMPDIR 供取消时精确清理，不误伤并发的其他备份 + tmpDir := fmt.Sprintf(`${TMPDIR:-/tmp}/ace-backup-task-%s`, str.Random(16)) + cmd := fmt.Sprintf(`%sexport TMPDIR="%s" +mkdir -p "$TMPDIR" +%s +rc=$? +rm -rf "$TMPDIR" +exit $rc`, pathEnv, tmpDir, backupCmd) + task := &biz.Task{ - Name: s.t.Get("Backup %s: %s", req.Type, req.Target), - Status: biz.TaskStatusWaiting, - Shell: cmd, + Key: fmt.Sprintf("backup:%s:%s", req.Type, req.Target), + Name: s.t.Get("Backup %s: %s", req.Type, req.Target), + Status: biz.TaskStatusWaiting, + Shell: cmd, + CancelShell: fmt.Sprintf(`rm -rf "%s"`, tmpDir), } if err = s.taskRepo.Push(task); err != nil { Error(w, http.StatusInternalServerError, "%v", err) diff --git a/internal/service/environment_php.go b/internal/service/environment_php.go index d72a6da2..debc7653 100644 --- a/internal/service/environment_php.go +++ b/internal/service/environment_php.go @@ -307,6 +307,7 @@ func (s *EnvironmentPHPService) InstallModule(w http.ResponseWriter, r *http.Req } task := new(biz.Task) + task.Key = fmt.Sprintf("php:module:%d:%s", req.Version, req.Slug) task.Name = s.t.Get("Install PHP-%d %s module", req.Version, req.Slug) task.Status = biz.TaskStatusWaiting task.Shell = cmd @@ -341,6 +342,7 @@ func (s *EnvironmentPHPService) UninstallModule(w http.ResponseWriter, r *http.R } task := new(biz.Task) + task.Key = fmt.Sprintf("php:module:%d:%s", req.Version, req.Slug) task.Name = s.t.Get("Uninstall PHP-%d %s module", req.Version, req.Slug) task.Status = biz.TaskStatusWaiting task.Shell = cmd diff --git a/internal/service/file.go b/internal/service/file.go index 715b1bc7..8f8d5004 100644 --- a/internal/service/file.go +++ b/internal/service/file.go @@ -457,9 +457,11 @@ func (s *FileService) RemoteDownload(w http.ResponseWriter, r *http.Request) { } task := new(biz.Task) + task.Key = "download:" + req.Path task.Name = s.t.Get("Download remote file %v", filepath.Base(req.Path)) task.Status = biz.TaskStatusWaiting task.Shell = fmt.Sprintf(`aria2c -c --file-allocation=falloc --allow-overwrite=true --auto-file-renaming=false --check-certificate=false --retry-wait=5 --max-tries=5 -x 16 -s 16 -k 1M -d '%s' -o '%s' '%s' && chmod 0755 '%s' && chown www:www '%s'`, filepath.Dir(req.Path), filepath.Base(req.Path), req.URL, req.Path, req.Path) + task.CancelShell = fmt.Sprintf(`rm -f '%s' '%s.aria2'`, req.Path, req.Path) if err = s.taskRepo.Push(task); err != nil { Error(w, http.StatusInternalServerError, "%v", err) @@ -585,9 +587,11 @@ func (s *FileService) Compress(w http.ResponseWriter, r *http.Request) { } task := new(biz.Task) + task.Key = "compress:" + req.File task.Name = s.t.Get("Compress %v", filepath.Base(req.File)) task.Status = biz.TaskStatusWaiting task.Shell = fmt.Sprintf(`%s && chmod 0755 '%s' && chown www:www '%s'`, cmd, req.File, req.File) + task.CancelShell = fmt.Sprintf(`rm -f '%s'`, req.File) if err = s.taskRepo.Push(task); err != nil { Error(w, http.StatusInternalServerError, "%v", err) @@ -611,6 +615,7 @@ func (s *FileService) UnCompress(w http.ResponseWriter, r *http.Request) { } task := new(biz.Task) + task.Key = fmt.Sprintf("uncompress:%s:%s", req.File, req.Path) task.Name = s.t.Get("Uncompress %v", filepath.Base(req.File)) task.Status = biz.TaskStatusWaiting task.Shell = fmt.Sprintf(`%s && chmod -R 0755 '%s' && chown -R www:www '%s'`, cmd, req.Path, req.Path) diff --git a/internal/service/task.go b/internal/service/task.go index 818112dd..7ab1e450 100644 --- a/internal/service/task.go +++ b/internal/service/task.go @@ -76,3 +76,18 @@ func (s *TaskService) Delete(w http.ResponseWriter, r *http.Request) { Success(w, nil) } + +func (s *TaskService) Cancel(w http.ResponseWriter, r *http.Request) { + req, err := Bind[request.ID](r) + if err != nil { + Error(w, http.StatusUnprocessableEntity, "%v", err) + return + } + + if err = s.taskRepo.Cancel(req.ID); err != nil { + Error(w, http.StatusInternalServerError, "%v", err) + return + } + + Success(w, nil) +} diff --git a/internal/taskqueue/runner.go b/internal/taskqueue/runner.go index a2aa7fb8..b060e796 100644 --- a/internal/taskqueue/runner.go +++ b/internal/taskqueue/runner.go @@ -5,7 +5,9 @@ import ( "fmt" "log/slog" "os" + "os/exec" "path/filepath" + "sync" "time" "gorm.io/gorm" @@ -19,6 +21,10 @@ type Runner struct { db *gorm.DB log *slog.Logger notify chan struct{} + + mu sync.Mutex + currentID uint // 当前运行的任务 ID + currentCancel context.CancelFunc // 取消当前任务 } // NewRunner 创建任务运行器 @@ -38,6 +44,17 @@ func (r *Runner) Notify() { } } +// Cancel 取消正在运行的任务，返回是否命中 +func (r *Runner) Cancel(id uint) bool { + r.mu.Lock() + defer r.mu.Unlock() + if r.currentID != id || r.currentCancel == nil { + return false + } + r.currentCancel() + return true +} + // Run 启动运行器 func (r *Runner) Run(ctx context.Context) { go func() { @@ -70,7 +87,7 @@ func (r *Runner) drain(ctx context.Context) { return default: } - if !r.processNext() { + if !r.processNext(ctx) { return } } @@ -84,20 +101,25 @@ func (r *Runner) clearZombie() { } // processNext 取一条 waiting 任务执行，返回是否有任务被处理 -func (r *Runner) processNext() bool { +func (r *Runner) processNext(ctx context.Context) bool { task := new(biz.Task) if err := r.db.Where("status = ?", biz.TaskStatusWaiting).Order("id asc").First(task).Error; err != nil { return false } - r.execute(task) + r.execute(ctx, task) return true } // execute 执行单个任务 -func (r *Runner) execute(task *biz.Task) { - if err := r.db.Model(task).Update("status", biz.TaskStatusRunning).Error; err != nil { - r.log.Error("failed to update task status to running", slog.Any("task_id", task.ID), slog.Any("err", err)) +func (r *Runner) execute(ctx context.Context, task *biz.Task) { + // 原子抢占，任务可能在取出后被取消 + result := r.db.Model(task).Where("status = ?", biz.TaskStatusWaiting).Update("status", biz.TaskStatusRunning) + if result.Error != nil { + r.log.Error("failed to update task status to running", slog.Any("task_id", task.ID), slog.Any("err", result.Error)) + return + } + if result.RowsAffected == 0 { return } @@ -110,9 +132,27 @@ func (r *Runner) execute(task *biz.Task) { return } - if err := shell.ExecWithLog(task.Shell, logFile); err != nil { - r.log.Warn("failed to execute background task", slog.Any("task_id", task.ID), slog.Any("err", err)) - _ = r.db.Model(task).Update("status", biz.TaskStatusFailed).Error + // 登记当前任务，供 Cancel 定位 + taskCtx, cancel := context.WithCancel(ctx) + r.mu.Lock() + r.currentID, r.currentCancel = task.ID, cancel + r.mu.Unlock() + defer func() { + r.mu.Lock() + r.currentID, r.currentCancel = 0, nil + r.mu.Unlock() + cancel() + }() + + if err := shell.ExecWithLog(taskCtx, task.Shell, logFile); err != nil { + // 用户取消标记为 canceled，面板停机保持 failed 由下次启动清理语义兜底 + status := biz.TaskStatusFailed + if taskCtx.Err() != nil && ctx.Err() == nil { + status = biz.TaskStatusCanceled + r.runCancelShell(task, logFile) + } + r.log.Warn("background task did not finish", slog.Any("task_id", task.ID), slog.Any("status", status), slog.Any("err", err)) + _ = r.db.Model(task).Update("status", status).Error return } @@ -120,3 +160,27 @@ func (r *Runner) execute(task *biz.Task) { r.log.Error("failed to update task status to success", slog.Any("task_id", task.ID), slog.Any("err", err)) } } + +// runCancelShell 任务被取消后执行清理命令，输出追加到任务日志 +func (r *Runner) runCancelShell(task *biz.Task, logFile string) { + if task.CancelShell == "" { + return + } + + cleanupCtx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + + f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) + if err != nil { + r.log.Warn("failed to open task log for cancel shell", slog.Any("task_id", task.ID), slog.Any("err", err)) + return + } + defer func(f *os.File) { _ = f.Close() }(f) + + cmd := exec.CommandContext(cleanupCtx, "bash", "-c", task.CancelShell) + cmd.Stdout = f + cmd.Stderr = f + if err = cmd.Run(); err != nil { + r.log.Warn("failed to run task cancel shell", slog.Any("task_id", task.ID), slog.Any("err", err)) + } +} diff --git a/internal/taskqueue/runner_test.go b/internal/taskqueue/runner_test.go new file mode 100644 index 00000000..286acb15 --- /dev/null +++ b/internal/taskqueue/runner_test.go @@ -0,0 +1,108 @@ +package taskqueue + +import ( + "log/slog" + "os" + "path/filepath" + "testing" + "time" + + "github.com/libtnb/sqlite" + "gorm.io/gorm" + + "github.com/acepanel/panel/v3/internal/biz" +) + +func newRunnerForTest(t *testing.T) *Runner { + t.Helper() + db, err := gorm.Open(sqlite.Open("file::memory:"), &gorm.Config{SkipDefaultTransaction: true}) + if err != nil { + t.Fatal(err) + } + if err = db.AutoMigrate(&biz.Task{}); err != nil { + t.Fatal(err) + } + return NewRunner(db, slog.New(slog.NewTextHandler(os.Stderr, nil))) +} + +// 等待任务进入指定状态 +func waitStatus(t *testing.T, db *gorm.DB, id uint, status biz.TaskStatus, timeout time.Duration) *biz.Task { + t.Helper() + deadline := time.Now().Add(timeout) + task := new(biz.Task) + for time.Now().Before(deadline) { + if err := db.First(task, id).Error; err == nil && task.Status == status { + return task + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("task %d did not reach status %s, current: %s", id, status, task.Status) + return nil +} + +func TestRunnerCancelRunning(t *testing.T) { + r := newRunnerForTest(t) + dir := t.TempDir() + marker := filepath.Join(dir, "cleanup.done") + + task := &biz.Task{ + Name: "sleep", + Status: biz.TaskStatusWaiting, + Shell: "sleep 60", + CancelShell: "touch " + marker, + } + if err := r.db.Create(task).Error; err != nil { + t.Fatal(err) + } + + r.Run(t.Context()) + + // 等待任务进入运行状态后取消 + waitStatus(t, r.db, task.ID, biz.TaskStatusRunning, 3*time.Second) + if !r.Cancel(task.ID) { + t.Fatal("Cancel should hit the running task") + } + + got := waitStatus(t, r.db, task.ID, biz.TaskStatusCanceled, 3*time.Second) + if got.Log == "" { + t.Fatal("task log path should be set") + } + + // 取消清理命令应已执行 + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if _, err := os.Stat(marker); err == nil { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("cancel shell was not executed") +} + +func TestRunnerCancelMiss(t *testing.T) { + r := newRunnerForTest(t) + if r.Cancel(1) { + t.Fatal("Cancel should miss when nothing is running") + } +} + +func TestRunnerWaitingCanceledNotExecuted(t *testing.T) { + r := newRunnerForTest(t) + + // 已被标记取消的任务不应被运行器抢占执行 + task := &biz.Task{Name: "noop", Status: biz.TaskStatusCanceled, Shell: "true"} + if err := r.db.Create(task).Error; err != nil { + t.Fatal(err) + } + + r.Run(t.Context()) + + time.Sleep(200 * time.Millisecond) + got := new(biz.Task) + if err := r.db.First(got, task.ID).Error; err != nil { + t.Fatal(err) + } + if got.Status != biz.TaskStatusCanceled { + t.Fatalf("canceled task should stay canceled, got %s", got.Status) + } +} diff --git a/mocks/biz/TaskRepo.go b/mocks/biz/TaskRepo.go index 12a556ce..f339d8dd 100644 --- a/mocks/biz/TaskRepo.go +++ b/mocks/biz/TaskRepo.go @@ -20,6 +20,52 @@ func (_m *TaskRepo) EXPECT() *TaskRepo_Expecter { return &TaskRepo_Expecter{mock: &_m.Mock} } +// Cancel provides a mock function with given fields: id +func (_m *TaskRepo) Cancel(id uint) error { + ret := _m.Called(id) + + if len(ret) == 0 { + panic("no return value specified for Cancel") + } + + var r0 error + if rf, ok := ret.Get(0).(func(uint) error); ok { + r0 = rf(id) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// TaskRepo_Cancel_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Cancel' +type TaskRepo_Cancel_Call struct { + *mock.Call +} + +// Cancel is a helper method to define mock.On call +// - id uint +func (_e *TaskRepo_Expecter) Cancel(id interface{}) *TaskRepo_Cancel_Call { + return &TaskRepo_Cancel_Call{Call: _e.mock.On("Cancel", id)} +} + +func (_c *TaskRepo_Cancel_Call) Run(run func(id uint)) *TaskRepo_Cancel_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(uint)) + }) + return _c +} + +func (_c *TaskRepo_Cancel_Call) Return(_a0 error) *TaskRepo_Cancel_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *TaskRepo_Cancel_Call) RunAndReturn(run func(uint) error) *TaskRepo_Cancel_Call { + _c.Call.Return(run) + return _c +} + // Delete provides a mock function with given fields: id func (_m *TaskRepo) Delete(id uint) error { ret := _m.Called(id) diff --git a/pkg/shell/exec.go b/pkg/shell/exec.go index 863f7872..d1938813 100644 --- a/pkg/shell/exec.go +++ b/pkg/shell/exec.go @@ -10,6 +10,7 @@ import ( "os/exec" "slices" "strings" + "syscall" "time" "github.com/creack/pty" @@ -32,7 +33,8 @@ func Exec(shell string) (string, error) { } // ExecWithLog 执行 shell 命令并将输出写入指定的日志文件 -func ExecWithLog(shell string, logFile string) error { +// ctx 取消时会杀死整个进程组 +func ExecWithLog(ctx context.Context, shell string, logFile string) error { _ = os.Setenv("LC_ALL", "C") f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) @@ -41,11 +43,19 @@ func ExecWithLog(shell string, logFile string) error { } defer func(f *os.File) { _ = f.Close() }(f) - cmd := exec.Command("bash", "-c", shell) + cmd := exec.CommandContext(ctx, "bash", "-c", shell) cmd.Stdout = f cmd.Stderr = f + // 命令会派生子进程（下载、压缩等），放入独立进程组以便取消时整组杀死 + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + cmd.Cancel = func() error { + return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + } if err = cmd.Run(); err != nil { + if ctx.Err() != nil { + return ctx.Err() + } return fmt.Errorf("run %s failed, err: %w", shell, err) } diff --git a/pkg/types/task.go b/pkg/types/task.go index 43d54c20..45d1cb76 100644 --- a/pkg/types/task.go +++ b/pkg/types/task.go @@ -6,4 +6,5 @@ import "context" type TaskRunner interface { Run(ctx context.Context) Notify() + Cancel(id uint) bool } diff --git a/web/src/api/panel/task/index.ts b/web/src/api/panel/task/index.ts index e7ed0df1..d303ff68 100644 --- a/web/src/api/panel/task/index.ts +++ b/web/src/api/panel/task/index.ts @@ -9,4 +9,6 @@ export default { get: (id: number): any => http.Get(`/task/${id}`), // 删除任务 delete: (id: number): any => http.Delete(`/task/${id}`), + // 取消任务 + cancel: (id: number): any => http.Post(`/task/${id}/cancel`), } diff --git a/web/src/views/task/TaskView.vue b/web/src/views/task/TaskView.vue index 0cb12658..cc232d43 100644 --- a/web/src/views/task/TaskView.vue +++ b/web/src/views/task/TaskView.vue @@ -9,7 +9,7 @@ import { useConfirm } from '@/components/system/composables/useConfirm' import { formatDateTime } from '@/utils' const { $gettext } = useGettext() -const { confirmDelete } = useConfirm() +const { confirmDelete, confirmAction } = useConfirm() const logModal = ref(false) const logPath = ref('') const logModalRef = ref<{ clear: () => void } | null>(null) @@ -42,7 +42,9 @@ const columns: any = [ ? $gettext('Waiting') : row.status === 'failed' ? $gettext('Failed') - : $gettext('Running') + : row.status === 'canceled' + ? $gettext('Canceled') + : $gettext('Running') }, }, { @@ -70,7 +72,7 @@ const columns: any = [ hideInExcel: true, render(row: any) { const items: any[] = [] - if (row.status != 'waiting') { + if (row.log) { items.push( h( NButton, @@ -87,7 +89,28 @@ const columns: any = [ ), ) } - if (row.status != 'waiting' && row.status != 'running') { + if (row.status == 'waiting' || row.status == 'running') { + items.push( + h( + NButton, + { + size: 'small', + type: 'error', + secondary: true, + onClick: async () => { + const ok = await confirmAction({ + title: $gettext('Cancel Task'), + content: $gettext('Are you sure you want to cancel task %{ name }?', { + name: row.name, + }), + }) + if (ok) handleCancel(row.id) + }, + }, + { default: () => $gettext('Cancel') }, + ), + ) + } else { items.push( h( NButton, @@ -127,6 +150,13 @@ const handleDelete = (id: number) => { }) } +const handleCancel = (id: number) => { + useRequest(task.cancel(id)).onSuccess(() => { + refresh() + window.$message.success($gettext('Canceled successfully')) + }) +} + onMounted(() => { refresh() })