chore: refactor for Actions Done Notification (go-gitea#7510)

Reviewed-on: https://codeberg.org/forgejo/forgejo/pulls/7510
Reviewed-by: Earl Warren <[email protected]>

Author: Earl Warren

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 {

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

models/actions/task.go

@@ -17,10 +17,8 @@ import (
 	"forgejo.org/modules/timeutil"
 	"forgejo.org/modules/util"
 
-	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/nektos/act/pkg/jobparser"
-	"google.golang.org/protobuf/types/known/timestamppb"
 	"xorm.io/builder"
 )
 
@@ -337,140 +335,6 @@ func UpdateTask(ctx context.Context, task *ActionTask, cols ...string) error {
 	return err
 }
 
-// UpdateTaskByState updates the task by the state.
-// It will always update the task if the state is not final, even there is no change.
-// So it will update ActionTask.Updated to avoid the task being judged as a zombie task.
-func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.TaskState) (*ActionTask, error) {
-	stepStates := map[int64]*runnerv1.StepState{}
-	for _, v := range state.Steps {
-		stepStates[v.Id] = v
-	}
-
-	ctx, commiter, err := db.TxContext(ctx)
-	if err != nil {
-		return nil, err
-	}
-	defer commiter.Close()
-
-	e := db.GetEngine(ctx)
-
-	task := &ActionTask{}
-	if has, err := e.ID(state.Id).Get(task); err != nil {
-		return nil, err
-	} else if !has {
-		return nil, util.ErrNotExist
-	} else if runnerID != task.RunnerID {
-		return nil, fmt.Errorf("invalid runner for task")
-	}
-
-	if task.Status.IsDone() {
-		// the state is final, do nothing
-		return task, nil
-	}
-
-	// state.Result is not unspecified means the task is finished
-	if state.Result != runnerv1.Result_RESULT_UNSPECIFIED {
-		task.Status = Status(state.Result)
-		task.Stopped = timeutil.TimeStamp(state.StoppedAt.AsTime().Unix())
-		if err := UpdateTask(ctx, task, "status", "stopped"); err != nil {
-			return nil, err
-		}
-		if _, err := UpdateRunJob(ctx, &ActionRunJob{
-			ID:      task.JobID,
-			Status:  task.Status,
-			Stopped: task.Stopped,
-		}, nil); err != nil {
-			return nil, err
-		}
-	} else {
-		// Force update ActionTask.Updated to avoid the task being judged as a zombie task
-		task.Updated = timeutil.TimeStampNow()
-		if err := UpdateTask(ctx, task, "updated"); err != nil {
-			return nil, err
-		}
-	}
-
-	if err := task.LoadAttributes(ctx); err != nil {
-		return nil, err
-	}
-
-	for _, step := range task.Steps {
-		var result runnerv1.Result
-		if v, ok := stepStates[step.Index]; ok {
-			result = v.Result
-			step.LogIndex = v.LogIndex
-			step.LogLength = v.LogLength
-			step.Started = convertTimestamp(v.StartedAt)
-			step.Stopped = convertTimestamp(v.StoppedAt)
-		}
-		if result != runnerv1.Result_RESULT_UNSPECIFIED {
-			step.Status = Status(result)
-		} else if step.Started != 0 {
-			step.Status = StatusRunning
-		}
-		if _, err := e.ID(step.ID).Update(step); err != nil {
-			return nil, err
-		}
-	}
-
-	if err := commiter.Commit(); err != nil {
-		return nil, err
-	}
-
-	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)
 
@@ -481,13 +345,6 @@ func FindOldTasksToExpire(ctx context.Context, olderThan timeutil.TimeStamp, lim
 		Find(&tasks)
 }
 
-func convertTimestamp(timestamp *timestamppb.Timestamp) timeutil.TimeStamp {
-	if timestamp.GetSeconds() == 0 && timestamp.GetNanos() == 0 {
-		return timeutil.TimeStamp(0)
-	}
-	return timeutil.TimeStamp(timestamp.AsTime().Unix())
-}
-
 func logFileName(repoFullName string, taskID int64) string {
 	ret := fmt.Sprintf("%s/%02x/%d.log", repoFullName, taskID%256, taskID)
 

routers/api/actions/runner/runner.go

@@ -178,7 +178,7 @@ func (s *Service) UpdateTask(
 ) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
 	runner := GetRunner(ctx)
 
-	task, err := actions_model.UpdateTaskByState(ctx, runner.ID, req.Msg.State)
+	task, err := actions_service.UpdateTaskByState(ctx, runner.ID, req.Msg.State)
 	if err != nil {
 		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("update task: %w", err))
 	}

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)

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
 			}
 		}

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)
 		}
 

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 {

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)
 		}
 	}