refactor: rename packages

2025-12-07 13:28:25 +00:00 · 2022-12-05 15:45:38 +08:00
parent fdd3444c52
commit 704f72017d
54 changed files with 81 additions and 81 deletions
@@ -0,0 +1,138 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package actions
+
+import (
+	"context"
+	"fmt"
+
+	bots_model "code.gitea.io/gitea/models/actions"
+	"code.gitea.io/gitea/models/db"
+	git_model "code.gitea.io/gitea/models/git"
+	repo_model "code.gitea.io/gitea/models/repo"
+	user_model "code.gitea.io/gitea/models/user"
+	"code.gitea.io/gitea/models/webhook"
+	bots_module "code.gitea.io/gitea/modules/actions"
+	"code.gitea.io/gitea/modules/graceful"
+	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/queue"
+	api "code.gitea.io/gitea/modules/structs"
+)
+
+func Init() {
+	jobEmitterQueue = queue.CreateUniqueQueue("bots_ready_job", jobEmitterQueueHandle, new(jobUpdate))
+	go graceful.GetManager().RunWithShutdownFns(jobEmitterQueue.Run)
+}
+
+func DeleteResourceOfRepository(ctx context.Context, repo *repo_model.Repository) error {
+	tasks, _, err := bots_model.FindTasks(ctx, bots_model.FindTaskOptions{RepoID: repo.ID})
+	if err != nil {
+		return fmt.Errorf("find task of repo %v: %w", repo.ID, err)
+	}
+
+	if err := db.WithTx(ctx, func(ctx context.Context) error {
+		e := db.GetEngine(ctx)
+		if _, err := e.Delete(&bots_model.BotTaskStep{RepoID: repo.ID}); err != nil {
+			return fmt.Errorf("delete bots task steps of repo %d: %w", repo.ID, err)
+		}
+		if _, err := e.Delete(&bots_model.BotTask{RepoID: repo.ID}); err != nil {
+			return fmt.Errorf("delete bots tasks of repo %d: %w", repo.ID, err)
+		}
+		if _, err := e.Delete(&bots_model.BotRunJob{RepoID: repo.ID}); err != nil {
+			return fmt.Errorf("delete bots run jobs of repo %d: %w", repo.ID, err)
+		}
+		if _, err := e.Delete(&bots_model.BotRun{RepoID: repo.ID}); err != nil {
+			return fmt.Errorf("delete bots runs of repo %d: %w", repo.ID, err)
+		}
+		if _, err := e.Delete(&bots_model.BotRunner{RepoID: repo.ID}); err != nil {
+			return fmt.Errorf("delete bots runner of repo %d: %w", repo.ID, err)
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	// remove logs file after tasks have been deleted, to avoid new log files
+	for _, task := range tasks {
+		err := bots_module.RemoveLogs(ctx, task.LogInStorage, task.LogFilename)
+		if err != nil {
+			log.Error("remove log file %q: %v", task.LogFilename, err)
+			// go on
+		}
+	}
+
+	return nil
+}
+
+func CreateCommitStatus(ctx context.Context, job *bots_model.BotRunJob) error {
+	if err := job.LoadAttributes(ctx); err != nil {
+		return fmt.Errorf("load run: %w", err)
+	}
+
+	run := job.Run
+	if run.Event != webhook.HookEventPush {
+		return nil
+	}
+
+	payload, err := run.GetPushEventPayload()
+	if err != nil {
+		return fmt.Errorf("GetPushEventPayload: %w", err)
+	}
+
+	creator, err := user_model.GetUserByID(ctx, payload.Pusher.ID)
+	if err != nil {
+		return fmt.Errorf("GetUserByID: %w", err)
+	}
+
+	repo := run.Repo
+	sha := payload.HeadCommit.ID
+	ctxname := job.Name
+	state := toCommitStatus(job.Status)
+
+	if statuses, _, err := git_model.GetLatestCommitStatus(ctx, repo.ID, sha, db.ListOptions{}); err == nil {
+		for _, v := range statuses {
+			if v.Context == ctxname {
+				if v.State == state {
+					return nil
+				}
+				break
+			}
+		}
+	} else {
+		return fmt.Errorf("GetLatestCommitStatus: %w", err)
+	}
+
+	if err := git_model.NewCommitStatus(git_model.NewCommitStatusOptions{
+		Repo:    repo,
+		SHA:     payload.HeadCommit.ID,
+		Creator: creator,
+		CommitStatus: &git_model.CommitStatus{
+			SHA:         sha,
+			TargetURL:   run.HTMLURL(),
+			Description: "",
+			Context:     ctxname,
+			CreatorID:   payload.Pusher.ID,
+			State:       state,
+		},
+	}); err != nil {
+		return fmt.Errorf("NewCommitStatus: %w", err)
+	}
+
+	return nil
+}
+
+func toCommitStatus(status bots_model.Status) api.CommitStatusState {
+	switch status {
+	case bots_model.StatusSuccess:
+		return api.CommitStatusSuccess
+	case bots_model.StatusFailure, bots_model.StatusCancelled, bots_model.StatusSkipped:
+		return api.CommitStatusFailure
+	case bots_model.StatusWaiting, bots_model.StatusBlocked:
+		return api.CommitStatusPending
+	case bots_model.StatusRunning:
+		return api.CommitStatusRunning
+	default:
+		return api.CommitStatusError
+	}
+}
@@ -0,0 +1,105 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package actions
+
+import (
+	"context"
+	"time"
+
+	bots_model "code.gitea.io/gitea/models/actions"
+	"code.gitea.io/gitea/models/db"
+	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/timeutil"
+)
+
+const (
+	zombieTaskTimeout   = 10 * time.Minute
+	endlessTaskTimeout  = 3 * time.Hour
+	abandonedJobTimeout = 24 * time.Hour
+)
+
+// StopZombieTasks stops the task which have running status, but haven't been updated for a long time
+func StopZombieTasks(ctx context.Context) error {
+	tasks, _, err := bots_model.FindTasks(ctx, bots_model.FindTaskOptions{
+		Status:        bots_model.StatusRunning,
+		UpdatedBefore: timeutil.TimeStamp(time.Now().Add(-zombieTaskTimeout).Unix()),
+	})
+	if err != nil {
+		log.Warn("find zombie tasks: %v", err)
+		return err
+	}
+
+	for _, task := range tasks {
+		if err := db.WithTx(ctx, func(ctx context.Context) error {
+			if err := bots_model.StopTask(ctx, task.ID, bots_model.StatusFailure); err != nil {
+				return err
+			}
+			if err := task.LoadJob(ctx); err != nil {
+				return err
+			}
+			return CreateCommitStatus(ctx, task.Job)
+		}); err != nil {
+			log.Warn("stop zombie task %v: %v", task.ID, err)
+			// go on
+		}
+	}
+	return nil
+}
+
+// StopEndlessTasks stops the tasks which have running status and continuous updates, but don't end for a long time
+func StopEndlessTasks(ctx context.Context) error {
+	tasks, _, err := bots_model.FindTasks(ctx, bots_model.FindTaskOptions{
+		Status:        bots_model.StatusRunning,
+		StartedBefore: timeutil.TimeStamp(time.Now().Add(-endlessTaskTimeout).Unix()),
+	})
+	if err != nil {
+		log.Warn("find endless tasks: %v", err)
+		return err
+	}
+
+	for _, task := range tasks {
+		if err := db.WithTx(ctx, func(ctx context.Context) error {
+			if err := bots_model.StopTask(ctx, task.ID, bots_model.StatusFailure); err != nil {
+				return err
+			}
+			if err := task.LoadJob(ctx); err != nil {
+				return err
+			}
+			return CreateCommitStatus(ctx, task.Job)
+		}); err != nil {
+			log.Warn("stop endless task %v: %v", task.ID, err)
+			// go on
+		}
+	}
+	return nil
+}
+
+// CancelAbandonedJobs cancels the jobs which have waiting status, but haven't been picked by a runner for a long time
+func CancelAbandonedJobs(ctx context.Context) error {
+	jobs, _, err := bots_model.FindRunJobs(ctx, bots_model.FindRunJobOptions{
+		Statuses:      []bots_model.Status{bots_model.StatusWaiting, bots_model.StatusBlocked},
+		UpdatedBefore: timeutil.TimeStamp(time.Now().Add(-abandonedJobTimeout).Unix()),
+	})
+	if err != nil {
+		log.Warn("find abandoned tasks: %v", err)
+		return err
+	}
+
+	now := timeutil.TimeStampNow()
+	for _, job := range jobs {
+		job.Status = bots_model.StatusCancelled
+		job.Stopped = now
+		if err := db.WithTx(ctx, func(ctx context.Context) error {
+			if _, err := bots_model.UpdateRunJob(ctx, job, nil, "status", "stopped"); err != nil {
+				return err
+			}
+			return CreateCommitStatus(ctx, job)
+		}); err != nil {
+			log.Warn("cancel abandoned job %v: %v", job.ID, err)
+			// go on
+		}
+	}
+
+	return nil
+}
@@ -0,0 +1,140 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package actions
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	bots_model "code.gitea.io/gitea/models/actions"
+	"code.gitea.io/gitea/models/db"
+	"code.gitea.io/gitea/modules/graceful"
+	"code.gitea.io/gitea/modules/queue"
+
+	"xorm.io/builder"
+)
+
+var jobEmitterQueue queue.UniqueQueue
+
+type jobUpdate struct {
+	RunID int64
+}
+
+func EmitJobsIfReady(runID int64) error {
+	err := jobEmitterQueue.Push(&jobUpdate{
+		RunID: runID,
+	})
+	if errors.Is(err, queue.ErrAlreadyInQueue) {
+		return nil
+	}
+	return err
+}
+
+func jobEmitterQueueHandle(data ...queue.Data) []queue.Data {
+	ctx := graceful.GetManager().ShutdownContext()
+	var ret []queue.Data
+	for _, d := range data {
+		update := d.(*jobUpdate)
+		if err := checkJobsOfRun(ctx, update.RunID); err != nil {
+			ret = append(ret, d)
+		}
+	}
+	return ret
+}
+
+func checkJobsOfRun(ctx context.Context, runID int64) error {
+	return db.WithTx(ctx, func(ctx context.Context) error {
+		jobs, _, err := bots_model.FindRunJobs(ctx, bots_model.FindRunJobOptions{RunID: runID})
+		if err != nil {
+			return err
+		}
+		idToJobs := make(map[string][]*bots_model.BotRunJob, len(jobs))
+		for _, job := range jobs {
+			idToJobs[job.JobID] = append(idToJobs[job.JobID], job)
+		}
+
+		updates := newJobStatusResolver(jobs).Resolve()
+		for _, job := range jobs {
+			if status, ok := updates[job.ID]; ok {
+				job.Status = status
+				if n, err := bots_model.UpdateRunJob(ctx, job, builder.Eq{"status": bots_model.StatusBlocked}, "status"); err != nil {
+					return err
+				} else if n != 1 {
+					return fmt.Errorf("no affected for updating blocked job %v", job.ID)
+				}
+			}
+		}
+		return nil
+	})
+}
+
+type jobStatusResolver struct {
+	statuses map[int64]bots_model.Status
+	needs    map[int64][]int64
+}
+
+func newJobStatusResolver(jobs bots_model.RunJobList) *jobStatusResolver {
+	idToJobs := make(map[string][]*bots_model.BotRunJob, len(jobs))
+	for _, job := range jobs {
+		idToJobs[job.JobID] = append(idToJobs[job.JobID], job)
+	}
+
+	statuses := make(map[int64]bots_model.Status, len(jobs))
+	needs := make(map[int64][]int64, len(jobs))
+	for _, job := range jobs {
+		statuses[job.ID] = job.Status
+		for _, need := range job.Needs {
+			for _, v := range idToJobs[need] {
+				needs[job.ID] = append(needs[job.ID], v.ID)
+			}
+		}
+	}
+	return &jobStatusResolver{
+		statuses: statuses,
+		needs:    needs,
+	}
+}
+
+func (r *jobStatusResolver) Resolve() map[int64]bots_model.Status {
+	ret := map[int64]bots_model.Status{}
+	for i := 0; i < len(r.statuses); i++ {
+		updated := r.resolve()
+		if len(updated) == 0 {
+			return ret
+		}
+		for k, v := range updated {
+			ret[k] = v
+			r.statuses[k] = v
+		}
+	}
+	return ret
+}
+
+func (r *jobStatusResolver) resolve() map[int64]bots_model.Status {
+	ret := map[int64]bots_model.Status{}
+	for id, status := range r.statuses {
+		if status != bots_model.StatusBlocked {
+			continue
+		}
+		allDone, allSucceed := true, true
+		for _, need := range r.needs[id] {
+			needStatus := r.statuses[need]
+			if !needStatus.IsDone() {
+				allDone = false
+			}
+			if needStatus.In(bots_model.StatusFailure, bots_model.StatusCancelled, bots_model.StatusSkipped) {
+				allSucceed = false
+			}
+		}
+		if allDone {
+			if allSucceed {
+				ret[id] = bots_model.StatusWaiting
+			} else {
+				ret[id] = bots_model.StatusSkipped
+			}
+		}
+	}
+	return ret
+}
@@ -0,0 +1,80 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package actions
+
+import (
+	"testing"
+
+	bots_model "code.gitea.io/gitea/models/actions"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_jobStatusResolver_Resolve(t *testing.T) {
+	tests := []struct {
+		name string
+		jobs bots_model.RunJobList
+		want map[int64]bots_model.Status
+	}{
+		{
+			name: "no blocked",
+			jobs: bots_model.RunJobList{
+				{ID: 1, JobID: "1", Status: bots_model.StatusWaiting, Needs: []string{}},
+				{ID: 2, JobID: "2", Status: bots_model.StatusWaiting, Needs: []string{}},
+				{ID: 3, JobID: "3", Status: bots_model.StatusWaiting, Needs: []string{}},
+			},
+			want: map[int64]bots_model.Status{},
+		},
+		{
+			name: "single blocked",
+			jobs: bots_model.RunJobList{
+				{ID: 1, JobID: "1", Status: bots_model.StatusSuccess, Needs: []string{}},
+				{ID: 2, JobID: "2", Status: bots_model.StatusBlocked, Needs: []string{"1"}},
+				{ID: 3, JobID: "3", Status: bots_model.StatusWaiting, Needs: []string{}},
+			},
+			want: map[int64]bots_model.Status{
+				2: bots_model.StatusWaiting,
+			},
+		},
+		{
+			name: "multiple blocked",
+			jobs: bots_model.RunJobList{
+				{ID: 1, JobID: "1", Status: bots_model.StatusSuccess, Needs: []string{}},
+				{ID: 2, JobID: "2", Status: bots_model.StatusBlocked, Needs: []string{"1"}},
+				{ID: 3, JobID: "3", Status: bots_model.StatusBlocked, Needs: []string{"1"}},
+			},
+			want: map[int64]bots_model.Status{
+				2: bots_model.StatusWaiting,
+				3: bots_model.StatusWaiting,
+			},
+		},
+		{
+			name: "chain blocked",
+			jobs: bots_model.RunJobList{
+				{ID: 1, JobID: "1", Status: bots_model.StatusFailure, Needs: []string{}},
+				{ID: 2, JobID: "2", Status: bots_model.StatusBlocked, Needs: []string{"1"}},
+				{ID: 3, JobID: "3", Status: bots_model.StatusBlocked, Needs: []string{"2"}},
+			},
+			want: map[int64]bots_model.Status{
+				2: bots_model.StatusSkipped,
+				3: bots_model.StatusSkipped,
+			},
+		},
+		{
+			name: "loop need",
+			jobs: bots_model.RunJobList{
+				{ID: 1, JobID: "1", Status: bots_model.StatusBlocked, Needs: []string{"3"}},
+				{ID: 2, JobID: "2", Status: bots_model.StatusBlocked, Needs: []string{"1"}},
+				{ID: 3, JobID: "3", Status: bots_model.StatusBlocked, Needs: []string{"2"}},
+			},
+			want: map[int64]bots_model.Status{},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := newJobStatusResolver(tt.jobs)
+			assert.Equal(t, tt.want, r.Resolve())
+		})
+	}
+}