From 81b5c7ca6f21ecb5b38194bb9a3a5a115690e46f Mon Sep 17 00:00:00 2001 From: christopher-besch Date: Tue, 8 Apr 2025 14:32:16 +0200 Subject: [PATCH] feat: move StopTask, CancelPreviousJobs and CleanRepoScheduleTasks to services/actions This enables all action run state changes (from a not done to a done state) to also send a notification. Moved these: - models/actions/task.go|423 col 6| func StopTask(ctx context.Context, taskID int64, status Status) error { - models/actions/run.go|190 col 6| func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error { - models/actions/schedule.go|122 col 6| func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository, cancelPreviousJobs bool) error { --- models/actions/run.go | 69 --------------------- models/actions/schedule.go | 22 ------- models/actions/task.go | 51 ---------------- routers/api/v1/repo/repo.go | 3 +- routers/web/repo/actions/view.go | 2 +- routers/web/repo/setting/setting.go | 3 +- services/actions/clear_tasks.go | 2 +- services/actions/notifier_helper.go | 6 +- services/actions/schedule_tasks.go | 93 ++++++++++++++++++++++++++++- services/actions/task.go | 53 ++++++++++++++++ services/repository/branch.go | 5 +- services/repository/setting.go | 3 +- 12 files changed, 156 insertions(+), 156 deletions(-) diff --git a/models/actions/run.go b/models/actions/run.go index 671177a892..c1e04071a8 100644 --- a/models/actions/run.go +++ b/models/actions/run.go @@ -185,75 +185,6 @@ func updateRepoRunsNumbers(ctx context.Context, repo *repo_model.Repository) err return err } -// CancelPreviousJobs cancels all previous jobs of the same repository, reference, workflow, and event. -// It's useful when a new run is triggered, and all previous runs needn't be continued anymore. -func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error { - // Find all runs in the specified repository, reference, and workflow with non-final status - runs, total, err := db.FindAndCount[ActionRun](ctx, FindRunOptions{ - RepoID: repoID, - Ref: ref, - WorkflowID: workflowID, - TriggerEvent: event, - Status: []Status{StatusRunning, StatusWaiting, StatusBlocked}, - }) - if err != nil { - return err - } - - // If there are no runs found, there's no need to proceed with cancellation, so return nil. - if total == 0 { - return nil - } - - // Iterate over each found run and cancel its associated jobs. - for _, run := range runs { - // Find all jobs associated with the current run. - jobs, err := db.Find[ActionRunJob](ctx, FindRunJobOptions{ - RunID: run.ID, - }) - if err != nil { - return err - } - - // Iterate over each job and attempt to cancel it. - for _, job := range jobs { - // Skip jobs that are already in a terminal state (completed, cancelled, etc.). - status := job.Status - if status.IsDone() { - continue - } - - // If the job has no associated task (probably an error), set its status to 'Cancelled' and stop it. - if job.TaskID == 0 { - job.Status = StatusCancelled - job.Stopped = timeutil.TimeStampNow() - - // Update the job's status and stopped time in the database. - n, err := UpdateRunJob(ctx, job, builder.Eq{"task_id": 0}, "status", "stopped") - if err != nil { - return err - } - - // If the update affected 0 rows, it means the job has changed in the meantime, so we need to try again. - if n == 0 { - return fmt.Errorf("job has changed, try again") - } - - // Continue with the next job. - continue - } - - // If the job has an associated task, try to stop the task, effectively cancelling the job. - if err := StopTask(ctx, job.TaskID, StatusCancelled); err != nil { - return err - } - } - } - - // Return nil to indicate successful cancellation of all running and waiting jobs. - return nil -} - // InsertRun inserts a run // The title will be cut off at 255 characters if it's longer than 255 characters. func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWorkflow) error { diff --git a/models/actions/schedule.go b/models/actions/schedule.go index 633582e017..cc68e9eafc 100644 --- a/models/actions/schedule.go +++ b/models/actions/schedule.go @@ -5,7 +5,6 @@ package actions import ( "context" - "fmt" "time" "forgejo.org/models/db" @@ -119,27 +118,6 @@ func DeleteScheduleTaskByRepo(ctx context.Context, id int64) error { return committer.Commit() } -func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository, cancelPreviousJobs bool) error { - // If actions disabled when there is schedule task, this will remove the outdated schedule tasks - // There is no other place we can do this because the app.ini will be changed manually - if err := DeleteScheduleTaskByRepo(ctx, repo.ID); err != nil { - return fmt.Errorf("DeleteCronTaskByRepo: %v", err) - } - if cancelPreviousJobs { - // cancel running cron jobs of this repository and delete old schedules - if err := CancelPreviousJobs( - ctx, - repo.ID, - repo.DefaultBranch, - "", - webhook_module.HookEventSchedule, - ); err != nil { - return fmt.Errorf("CancelPreviousJobs: %v", err) - } - } - return nil -} - type FindScheduleOptions struct { db.ListOptions RepoID int64 diff --git a/models/actions/task.go b/models/actions/task.go index 63cbc6e586..db18eaa0b3 100644 --- a/models/actions/task.go +++ b/models/actions/task.go @@ -420,57 +420,6 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task return task, nil } -func StopTask(ctx context.Context, taskID int64, status Status) error { - if !status.IsDone() { - return fmt.Errorf("cannot stop task with status %v", status) - } - e := db.GetEngine(ctx) - - task := &ActionTask{} - if has, err := e.ID(taskID).Get(task); err != nil { - return err - } else if !has { - return util.ErrNotExist - } - if task.Status.IsDone() { - return nil - } - - now := timeutil.TimeStampNow() - task.Status = status - task.Stopped = now - if _, err := UpdateRunJob(ctx, &ActionRunJob{ - ID: task.JobID, - Status: task.Status, - Stopped: task.Stopped, - }, nil); err != nil { - return err - } - - if err := UpdateTask(ctx, task, "status", "stopped"); err != nil { - return err - } - - if err := task.LoadAttributes(ctx); err != nil { - return err - } - - for _, step := range task.Steps { - if !step.Status.IsDone() { - step.Status = status - if step.Started == 0 { - step.Started = now - } - step.Stopped = now - } - if _, err := e.ID(step.ID).Update(step); err != nil { - return err - } - } - - return nil -} - func FindOldTasksToExpire(ctx context.Context, olderThan timeutil.TimeStamp, limit int) ([]*ActionTask, error) { e := db.GetEngine(ctx) diff --git a/routers/api/v1/repo/repo.go b/routers/api/v1/repo/repo.go index 59b32eb667..50f628e035 100644 --- a/routers/api/v1/repo/repo.go +++ b/routers/api/v1/repo/repo.go @@ -11,7 +11,6 @@ import ( "strings" "time" - actions_model "forgejo.org/models/actions" activities_model "forgejo.org/models/activities" "forgejo.org/models/db" "forgejo.org/models/organization" @@ -1065,7 +1064,7 @@ func updateRepoArchivedState(ctx *context.APIContext, opts api.EditRepoOption) e ctx.Error(http.StatusInternalServerError, "ArchiveRepoState", err) return err } - if err := actions_model.CleanRepoScheduleTasks(ctx, repo, true); err != nil { + if err := actions_service.CleanRepoScheduleTasks(ctx, repo, true); err != nil { log.Error("CleanRepoScheduleTasks for archived repo %s/%s: %v", ctx.Repo.Owner.Name, repo.Name, err) } log.Trace("Repository was archived: %s/%s", ctx.Repo.Owner.Name, repo.Name) diff --git a/routers/web/repo/actions/view.go b/routers/web/repo/actions/view.go index 2d009c5720..563832297d 100644 --- a/routers/web/repo/actions/view.go +++ b/routers/web/repo/actions/view.go @@ -521,7 +521,7 @@ func Cancel(ctx *context_module.Context) { } continue } - if err := actions_model.StopTask(ctx, job.TaskID, actions_model.StatusCancelled); err != nil { + if err := actions_service.StopTask(ctx, job.TaskID, actions_model.StatusCancelled); err != nil { return err } } diff --git a/routers/web/repo/setting/setting.go b/routers/web/repo/setting/setting.go index 083cc4ae82..6e4f0f4829 100644 --- a/routers/web/repo/setting/setting.go +++ b/routers/web/repo/setting/setting.go @@ -14,7 +14,6 @@ import ( "time" "forgejo.org/models" - actions_model "forgejo.org/models/actions" "forgejo.org/models/db" "forgejo.org/models/organization" quota_model "forgejo.org/models/quota" @@ -1034,7 +1033,7 @@ func SettingsPost(ctx *context.Context) { return } - if err := actions_model.CleanRepoScheduleTasks(ctx, repo, true); err != nil { + if err := actions_service.CleanRepoScheduleTasks(ctx, repo, true); err != nil { log.Error("CleanRepoScheduleTasks for archived repo %s/%s: %v", ctx.Repo.Owner.Name, repo.Name, err) } diff --git a/services/actions/clear_tasks.go b/services/actions/clear_tasks.go index 31e15ec927..f34c07fe5c 100644 --- a/services/actions/clear_tasks.go +++ b/services/actions/clear_tasks.go @@ -41,7 +41,7 @@ func stopTasks(ctx context.Context, opts actions_model.FindTaskOptions) error { jobs := make([]*actions_model.ActionRunJob, 0, len(tasks)) for _, task := range tasks { if err := db.WithTx(ctx, func(ctx context.Context) error { - if err := actions_model.StopTask(ctx, task.ID, actions_model.StatusFailure); err != nil { + if err := StopTask(ctx, task.ID, actions_model.StatusFailure); err != nil { return err } if err := task.LoadJob(ctx); err != nil { diff --git a/services/actions/notifier_helper.go b/services/actions/notifier_helper.go index 9de0b75ac7..e55714a2c3 100644 --- a/services/actions/notifier_helper.go +++ b/services/actions/notifier_helper.go @@ -139,7 +139,7 @@ func notify(ctx context.Context, input *notifyInput) error { return nil } if unit_model.TypeActions.UnitGlobalDisabled() { - if err := actions_model.CleanRepoScheduleTasks(ctx, input.Repo, true); err != nil { + if err := CleanRepoScheduleTasks(ctx, input.Repo, true); err != nil { log.Error("CleanRepoScheduleTasks: %v", err) } return nil @@ -373,7 +373,7 @@ func handleWorkflows( // cancel running jobs if the event is push or pull_request_sync if run.Event == webhook_module.HookEventPush || run.Event == webhook_module.HookEventPullRequestSync { - if err := actions_model.CancelPreviousJobs( + if err := CancelPreviousJobs( ctx, run.RepoID, run.Ref, @@ -504,7 +504,7 @@ func handleSchedules( log.Error("CountSchedules: %v", err) return err } else if count > 0 { - if err := actions_model.CleanRepoScheduleTasks(ctx, input.Repo, false); err != nil { + if err := CleanRepoScheduleTasks(ctx, input.Repo, false); err != nil { log.Error("CleanRepoScheduleTasks: %v", err) } } diff --git a/services/actions/schedule_tasks.go b/services/actions/schedule_tasks.go index f66a6ca092..8d41797a0e 100644 --- a/services/actions/schedule_tasks.go +++ b/services/actions/schedule_tasks.go @@ -17,6 +17,7 @@ import ( webhook_module "forgejo.org/modules/webhook" "github.com/nektos/act/pkg/jobparser" + "xorm.io/builder" ) // StartScheduleTasks start the task @@ -55,7 +56,7 @@ func startTasks(ctx context.Context) error { // cancel running jobs if the event is push if row.Schedule.Event == webhook_module.HookEventPush { // cancel running jobs of the same workflow - if err := actions_model.CancelPreviousJobs( + if err := CancelPreviousJobs( ctx, row.RepoID, row.Schedule.Ref, @@ -152,3 +153,93 @@ func CreateScheduleTask(ctx context.Context, cron *actions_model.ActionSchedule) // Return nil if no errors occurred return nil } + +// CancelPreviousJobs cancels all previous jobs of the same repository, reference, workflow, and event. +// It's useful when a new run is triggered, and all previous runs needn't be continued anymore. +func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error { + // Find all runs in the specified repository, reference, and workflow with non-final status + runs, total, err := db.FindAndCount[actions_model.ActionRun](ctx, actions_model.FindRunOptions{ + RepoID: repoID, + Ref: ref, + WorkflowID: workflowID, + TriggerEvent: event, + Status: []actions_model.Status{actions_model.StatusRunning, actions_model.StatusWaiting, actions_model.StatusBlocked}, + }) + if err != nil { + return err + } + + // If there are no runs found, there's no need to proceed with cancellation, so return nil. + if total == 0 { + return nil + } + + // Iterate over each found run and cancel its associated jobs. + for _, run := range runs { + // Find all jobs associated with the current run. + jobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{ + RunID: run.ID, + }) + if err != nil { + return err + } + + // Iterate over each job and attempt to cancel it. + for _, job := range jobs { + // Skip jobs that are already in a terminal state (completed, cancelled, etc.). + status := job.Status + if status.IsDone() { + continue + } + + // If the job has no associated task (probably an error), set its status to 'Cancelled' and stop it. + if job.TaskID == 0 { + job.Status = actions_model.StatusCancelled + job.Stopped = timeutil.TimeStampNow() + + // Update the job's status and stopped time in the database. + n, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"task_id": 0}, "status", "stopped") + if err != nil { + return err + } + + // If the update affected 0 rows, it means the job has changed in the meantime, so we need to try again. + if n == 0 { + return fmt.Errorf("job has changed, try again") + } + + // Continue with the next job. + continue + } + + // If the job has an associated task, try to stop the task, effectively cancelling the job. + if err := StopTask(ctx, job.TaskID, actions_model.StatusCancelled); err != nil { + return err + } + } + } + + // Return nil to indicate successful cancellation of all running and waiting jobs. + return nil +} + +func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository, cancelPreviousJobs bool) error { + // If actions disabled when there is schedule task, this will remove the outdated schedule tasks + // There is no other place we can do this because the app.ini will be changed manually + if err := actions_model.DeleteScheduleTaskByRepo(ctx, repo.ID); err != nil { + return fmt.Errorf("DeleteCronTaskByRepo: %v", err) + } + if cancelPreviousJobs { + // cancel running cron jobs of this repository and delete old schedules + if err := CancelPreviousJobs( + ctx, + repo.ID, + repo.DefaultBranch, + "", + webhook_module.HookEventSchedule, + ); err != nil { + return fmt.Errorf("CancelPreviousJobs: %v", err) + } + } + return nil +} diff --git a/services/actions/task.go b/services/actions/task.go index 43c8deaa5f..40bf4a8e42 100644 --- a/services/actions/task.go +++ b/services/actions/task.go @@ -10,6 +10,8 @@ import ( actions_model "forgejo.org/models/actions" "forgejo.org/models/db" secret_model "forgejo.org/models/secret" + "forgejo.org/modules/timeutil" + "forgejo.org/modules/util" runnerv1 "code.gitea.io/actions-proto-go/runner/v1" "google.golang.org/protobuf/types/known/structpb" @@ -105,3 +107,54 @@ func findTaskNeeds(ctx context.Context, taskJob *actions_model.ActionRunJob) (ma } return ret, nil } + +func StopTask(ctx context.Context, taskID int64, status actions_model.Status) error { + if !status.IsDone() { + return fmt.Errorf("cannot stop task with status %v", status) + } + e := db.GetEngine(ctx) + + task := &actions_model.ActionTask{} + if has, err := e.ID(taskID).Get(task); err != nil { + return err + } else if !has { + return util.ErrNotExist + } + if task.Status.IsDone() { + return nil + } + + now := timeutil.TimeStampNow() + task.Status = status + task.Stopped = now + if _, err := actions_model.UpdateRunJob(ctx, &actions_model.ActionRunJob{ + ID: task.JobID, + Status: task.Status, + Stopped: task.Stopped, + }, nil); err != nil { + return err + } + + if err := actions_model.UpdateTask(ctx, task, "status", "stopped"); err != nil { + return err + } + + if err := task.LoadAttributes(ctx); err != nil { + return err + } + + for _, step := range task.Steps { + if !step.Status.IsDone() { + step.Status = status + if step.Started == 0 { + step.Started = now + } + step.Stopped = now + } + if _, err := e.ID(step.ID).Update(step); err != nil { + return err + } + } + + return nil +} diff --git a/services/repository/branch.go b/services/repository/branch.go index 689f35803d..d5b68c1fa7 100644 --- a/services/repository/branch.go +++ b/services/repository/branch.go @@ -28,6 +28,7 @@ import ( "forgejo.org/modules/timeutil" "forgejo.org/modules/util" webhook_module "forgejo.org/modules/webhook" + actions_service "forgejo.org/services/actions" notify_service "forgejo.org/services/notify" pull_service "forgejo.org/services/pull" files_service "forgejo.org/services/repository/files" @@ -377,7 +378,7 @@ func RenameBranch(ctx context.Context, repo *repo_model.Repository, doer *user_m log.Error("DeleteCronTaskByRepo: %v", err) } // cancel running cron jobs of this repository and delete old schedules - if err := actions_model.CancelPreviousJobs( + if err := actions_service.CancelPreviousJobs( ctx, repo.ID, from, @@ -578,7 +579,7 @@ func SetRepoDefaultBranch(ctx context.Context, repo *repo_model.Repository, gitR log.Error("DeleteCronTaskByRepo: %v", err) } // cancel running cron jobs of this repository and delete old schedules - if err := actions_model.CancelPreviousJobs( + if err := actions_service.CancelPreviousJobs( ctx, repo.ID, oldDefaultBranchName, diff --git a/services/repository/setting.go b/services/repository/setting.go index c127f3129e..68cdfc370b 100644 --- a/services/repository/setting.go +++ b/services/repository/setting.go @@ -7,7 +7,6 @@ import ( "context" "slices" - actions_model "forgejo.org/models/actions" "forgejo.org/models/db" repo_model "forgejo.org/models/repo" "forgejo.org/models/unit" @@ -29,7 +28,7 @@ func UpdateRepositoryUnits(ctx context.Context, repo *repo_model.Repository, uni } if slices.Contains(deleteUnitTypes, unit.TypeActions) { - if err := actions_model.CleanRepoScheduleTasks(ctx, repo, true); err != nil { + if err := actions_service.CleanRepoScheduleTasks(ctx, repo, true); err != nil { log.Error("CleanRepoScheduleTasks: %v", err) } }