luckyimnotanigger/forgejo
1
0
Fork 0
mirror of https://codeberg.org/forgejo/forgejo.git synced 2025-04-16 12:12:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

chore: refactor for Actions Done Notification (#7510)

Browse source 
Reviewed-on: https://codeberg.org/forgejo/forgejo/pulls/7510
Reviewed-by: Earl Warren <earl-warren@noreply.codeberg.org>
This commit is contained in:
Earl Warren 2025-04-10 09:51:57 +00:00
commit a783a72d6b
13 changed files with 248 additions and 249 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
models/actions/run.go
View file

@ -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 {

22
models/actions/schedule.go
View file

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

143
models/actions/task.go
View file

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

2
routers/api/actions/runner/runner.go
View file

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

3
routers/api/v1/repo/repo.go
View file

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

2
routers/web/repo/actions/view.go
View file

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

3
routers/web/repo/setting/setting.go
View file

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

2
services/actions/clear_tasks.go
View file

@ -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 {

6
services/actions/notifier_helper.go
View file

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

93
services/actions/schedule_tasks.go
View file

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

144
services/actions/task.go
View file

@ -10,9 +10,12 @@ 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"
"google.golang.org/protobuf/types/known/timestamppb"
)
func PickTask(ctx context.Context, runner *actions_model.ActionRunner) (*runnerv1.Task, bool, error) {
@ -105,3 +108,144 @@ 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
}
// 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())
}

5
services/repository/branch.go
View file

@ -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,

3
services/repository/setting.go
View file

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