From cdb4682bca34a78fc9f70a8195f9dd4f16e61fa4 Mon Sep 17 00:00:00 2001 From: christopher-besch Date: Tue, 8 Apr 2025 17:41:13 +0200 Subject: [PATCH] feat: move UpdateTaskByState to services This function is also can't be in models in order to enable calling the action run state change notification channel. --- models/actions/task.go | 92 ---------------------------- routers/api/actions/runner/runner.go | 2 +- services/actions/task.go | 91 +++++++++++++++++++++++++++ 3 files changed, 92 insertions(+), 93 deletions(-) diff --git a/models/actions/task.go b/models/actions/task.go index db18eaa0b3..c68b5b97ff 100644 --- a/models/actions/task.go +++ b/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,89 +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 FindOldTasksToExpire(ctx context.Context, olderThan timeutil.TimeStamp, limit int) ([]*ActionTask, error) { e := db.GetEngine(ctx) @@ -430,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) diff --git a/routers/api/actions/runner/runner.go b/routers/api/actions/runner/runner.go index 8aae69f463..a971cd3fbf 100644 --- a/routers/api/actions/runner/runner.go +++ b/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)) } diff --git a/services/actions/task.go b/services/actions/task.go index 40bf4a8e42..6d89afed45 100644 --- a/services/actions/task.go +++ b/services/actions/task.go @@ -15,6 +15,7 @@ import ( runnerv1 "code.gitea.io/actions-proto-go/runner/v1" "google.golang.org/protobuf/types/known/structpb" + "google.golang.org/protobuf/types/known/timestamppb" ) func PickTask(ctx context.Context, runner *actions_model.ActionRunner) (*runnerv1.Task, bool, error) { @@ -158,3 +159,93 @@ func StopTask(ctx context.Context, taskID int64, status actions_model.Status) er return nil } + +// 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) (*actions_model.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 := &actions_model.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 = actions_model.Status(state.Result) + task.Stopped = timeutil.TimeStamp(state.StoppedAt.AsTime().Unix()) + if err := actions_model.UpdateTask(ctx, task, "status", "stopped"); err != nil { + return nil, err + } + if _, err := actions_model.UpdateRunJob(ctx, &actions_model.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 := actions_model.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 = actions_model.Status(result) + } else if step.Started != 0 { + step.Status = actions_model.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 convertTimestamp(timestamp *timestamppb.Timestamp) timeutil.TimeStamp { + if timestamp.GetSeconds() == 0 && timestamp.GetNanos() == 0 { + return timeutil.TimeStamp(0) + } + return timeutil.TimeStamp(timestamp.AsTime().Unix()) +}