mirror of
https://github.com/go-gitea/gitea.git
synced 2026-10-21 09:06:46 +09:00
refactor(actions): small cleanups across models, services and routers (#39722)
Collection of small, behaviour-preserving cleanups in the Actions code. - Fix log and error prefixes that named non-existent or wrong functions, and wrap errors with `%w` so `errors.Is` keeps working. - Remove unused code. - Rely on `ctx.ServerError` / `ctx.APIErrorAuto`, which already map `util.ErrNotExist` and friends, instead of equivalent hand-written switches. - Load runs referenced by commit statuses with one query instead of one per run (used on every PR view); the result order is now stable. - Encode an event's payload once instead of up to three times. - Share the duplicated runner-token "get or create" logic and the "notify and emit" call pair. - Simplify `ToActionArtifact` / `ToActionRunner` signatures. --------- Signed-off-by: bircni <bircni@icloud.com> Co-authored-by: delvh <dev.lh@web.de> Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>
This commit is contained in:
26 files changed
+113
-223
No files matched your search
@@ -50,15 +50,13 @@ func stopTasksByStatuses(ctx context.Context, opts actions_model.FindTaskOptions
|
||||
|
||||
func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID string, event webhook_module.HookEventType) error {
|
||||
jobs, err := actions_model.CancelPreviousJobs(ctx, repoID, ref, workflowID, event)
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, jobs)
|
||||
EmitJobsIfReadyByJobs(jobs)
|
||||
notifyAndEmitFinishedJobs(ctx, jobs)
|
||||
return err
|
||||
}
|
||||
|
||||
func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository) error {
|
||||
jobs, err := actions_model.CleanRepoScheduleTasks(ctx, repo)
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, jobs)
|
||||
EmitJobsIfReadyByJobs(jobs)
|
||||
notifyAndEmitFinishedJobs(ctx, jobs)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -171,8 +169,7 @@ func stopTasks(ctx context.Context, opts actions_model.FindTaskOptions) error {
|
||||
remove()
|
||||
}
|
||||
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, jobs)
|
||||
EmitJobsIfReadyByJobs(jobs)
|
||||
notifyAndEmitFinishedJobs(ctx, jobs)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -193,8 +190,7 @@ func CancelAbandonedJobs(ctx context.Context) error {
|
||||
log.Warn("cancel abandoned jobs: %v", err)
|
||||
}
|
||||
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, updatedJobs)
|
||||
EmitJobsIfReadyByJobs(updatedJobs)
|
||||
notifyAndEmitFinishedJobs(ctx, updatedJobs)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -17,10 +17,10 @@ import (
|
||||
actions_module "gitea.dev/modules/actions"
|
||||
"gitea.dev/modules/actions/jobparser"
|
||||
"gitea.dev/modules/commitstatus"
|
||||
"gitea.dev/modules/container"
|
||||
"gitea.dev/modules/glob"
|
||||
"gitea.dev/modules/log"
|
||||
api "gitea.dev/modules/structs"
|
||||
"gitea.dev/modules/util"
|
||||
webhook_module "gitea.dev/modules/webhook"
|
||||
pull_service "gitea.dev/services/pull"
|
||||
commitstatus_service "gitea.dev/services/repository/commitstatus"
|
||||
@@ -36,7 +36,7 @@ func CreateCommitStatusForRunJobs(ctx context.Context, run *actions_model.Action
|
||||
|
||||
event, commitID, err := getCommitStatusEventNameAndCommitID(run)
|
||||
if err != nil {
|
||||
log.Error("GetCommitStatusEventNameAndSHA: %v", err)
|
||||
log.Error("getCommitStatusEventNameAndCommitID: %v", err)
|
||||
}
|
||||
if event == "" || commitID == "" {
|
||||
return // unsupported event, or no commit id, or error occurs, do nothing
|
||||
@@ -71,31 +71,15 @@ func CreateCommitStatusForRunJobs(ctx context.Context, run *actions_model.Action
|
||||
}
|
||||
}
|
||||
|
||||
func GetRunsFromCommitStatuses(ctx context.Context, statuses []*git_model.CommitStatus) ([]*actions_model.ActionRun, error) {
|
||||
runMap := make(map[int64]*actions_model.ActionRun)
|
||||
func GetRunsFromCommitStatuses(ctx context.Context, repoID int64, statuses []*git_model.CommitStatus) (runs []*actions_model.ActionRun, _ error) {
|
||||
runIDs := container.Set[int64]{}
|
||||
for _, status := range statuses {
|
||||
runID, _, ok := status.ParseGiteaActionsTargetURL(ctx)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
_, ok = runMap[runID]
|
||||
if !ok {
|
||||
run, err := actions_model.GetRunByRepoAndID(ctx, status.RepoID, runID)
|
||||
if err != nil {
|
||||
if errors.Is(err, util.ErrNotExist) {
|
||||
// the run may be deleted manually, just skip it
|
||||
continue
|
||||
}
|
||||
return nil, fmt.Errorf("GetRunByRepoAndID: %w", err)
|
||||
}
|
||||
runMap[runID] = run
|
||||
if ok {
|
||||
runIDs.Add(runID)
|
||||
}
|
||||
}
|
||||
runs := make([]*actions_model.ActionRun, 0, len(runMap))
|
||||
for _, run := range runMap {
|
||||
runs = append(runs, run)
|
||||
}
|
||||
return runs, nil
|
||||
return actions_model.GetRunsByRepoAndID(ctx, repoID, runIDs.Values())
|
||||
}
|
||||
|
||||
func getCommitStatusEventNameAndCommitID(run *actions_model.ActionRun) (event, commitID string, _ error) {
|
||||
|
||||
@@ -12,7 +12,6 @@ import (
|
||||
"gitea.dev/models/unit"
|
||||
"gitea.dev/modules/commitstatus"
|
||||
"gitea.dev/modules/git"
|
||||
"gitea.dev/modules/json"
|
||||
"gitea.dev/modules/log"
|
||||
"gitea.dev/modules/timeutil"
|
||||
"gitea.dev/modules/util"
|
||||
@@ -22,7 +21,7 @@ func handleInvalidWorkflows(ctx context.Context, input *notifyInput, ref git.Ref
|
||||
if len(invalid) == 0 {
|
||||
return
|
||||
}
|
||||
payload, err := json.Marshal(input.Payload)
|
||||
payload, err := input.EncodedPayload()
|
||||
if err != nil {
|
||||
log.Error("marshal event payload: %v", err)
|
||||
return
|
||||
@@ -35,7 +34,7 @@ func handleInvalidWorkflows(ctx context.Context, input *notifyInput, ref git.Ref
|
||||
insertInvalidWorkflowRun(ctx, &actions_model.ActionRun{
|
||||
Title: commit.MessageTitle(), RepoID: input.Repo.ID, Repo: input.Repo, OwnerID: input.Repo.OwnerID,
|
||||
WorkflowID: entryName, TriggerUserID: input.Doer.ID, TriggerUser: input.Doer, Ref: ref.String(),
|
||||
CommitSHA: commit.ID.String(), Event: input.Event, TriggerEvent: string(input.Event), EventPayload: string(payload),
|
||||
CommitSHA: commit.ID.String(), Event: input.Event, TriggerEvent: string(input.Event), EventPayload: payload,
|
||||
WorkflowRepoID: input.Repo.ID, WorkflowCommitSHA: commit.ID.String(),
|
||||
}, parseErr)
|
||||
}
|
||||
|
||||
@@ -103,8 +103,7 @@ func checkJobsByRunID(ctx context.Context, runID int64) error {
|
||||
log.Error("re-emit run %d: %v", rid, err)
|
||||
}
|
||||
}
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, result.CancelledJobs)
|
||||
EmitJobsIfReadyByJobs(result.CancelledJobs)
|
||||
notifyAndEmitFinishedJobs(ctx, result.CancelledJobs)
|
||||
if err := createCommitStatusesForJobsByRun(ctx, result.Jobs); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -651,7 +650,7 @@ func (r *jobStatusResolver) resolve(ctx context.Context) (map[int64]actions_mode
|
||||
|
||||
newStatus, cancelledJobs, err := PrepareToStartJobWithConcurrency(ctx, actionRunJob)
|
||||
if err != nil {
|
||||
log.Error("ShouldBlockJobByConcurrency failed, this job will stay blocked: job: %d, err: %v", id, err)
|
||||
log.Error("PrepareToStartJobWithConcurrency failed, this job will stay blocked: job: %d, err: %v", id, err)
|
||||
} else {
|
||||
r.cancelledJobs = append(r.cancelledJobs, cancelledJobs...)
|
||||
for _, cancelled := range cancelledJobs {
|
||||
|
||||
@@ -56,7 +56,7 @@ func (n *actionsNotifier) NewIssue(ctx context.Context, issue *issues_model.Issu
|
||||
Issue: convert.ToAPIIssue(ctx, issue.Poster, issue),
|
||||
Repository: convert.ToRepo(ctx, issue.Repo, permission),
|
||||
Sender: convert.ToUser(ctx, issue.Poster, nil),
|
||||
}).Notify(withMethod(ctx, "NewIssue"))
|
||||
}).Notify(ctx)
|
||||
}
|
||||
|
||||
// IssueChangeContent notifies change content of issue
|
||||
@@ -455,7 +455,7 @@ func (n *actionsNotifier) PullRequestReview(ctx context.Context, pr *issues_mode
|
||||
|
||||
permission, err := access_model.GetIndividualUserRepoPermission(ctx, review.Issue.Repo, review.Issue.Poster)
|
||||
if err != nil {
|
||||
log.Error("models.GetIndividualUserRepoPermission: %v", err)
|
||||
log.Error("GetIndividualUserRepoPermission: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -516,7 +516,7 @@ func (*actionsNotifier) MergePullRequest(ctx context.Context, doer *user_model.U
|
||||
}
|
||||
|
||||
if err := pr.LoadIssue(ctx); err != nil {
|
||||
log.Error("LoadAttributes: %v", err)
|
||||
log.Error("pr.LoadIssue: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -689,7 +689,7 @@ func (n *actionsNotifier) PullRequestSynchronized(ctx context.Context, doer *use
|
||||
ctx = withMethod(ctx, "PullRequestSynchronized")
|
||||
|
||||
if err := pr.LoadIssue(ctx); err != nil {
|
||||
log.Error("LoadAttributes: %v", err)
|
||||
log.Error("pr.LoadIssue: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -716,7 +716,7 @@ func (n *actionsNotifier) PullRequestChangeTargetBranch(ctx context.Context, doe
|
||||
ctx = withMethod(ctx, "PullRequestChangeTargetBranch")
|
||||
|
||||
if err := pr.LoadIssue(ctx); err != nil {
|
||||
log.Error("LoadAttributes: %v", err)
|
||||
log.Error("pr.LoadIssue: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -67,6 +67,8 @@ type notifyInput struct {
|
||||
Ref git.RefName
|
||||
Payload api.Payloader
|
||||
PullRequest *issues_model.PullRequest
|
||||
|
||||
encodedPayload string // cached by EncodedPayload
|
||||
}
|
||||
|
||||
func newNotifyInput(repo *repo_model.Repository, doer *user_model.User, event webhook_module.HookEventType) *notifyInput {
|
||||
@@ -111,6 +113,18 @@ func (input *notifyInput) WithPullRequest(pr *issues_model.PullRequest) *notifyI
|
||||
return input
|
||||
}
|
||||
|
||||
// EncodedPayload marshals Payload once, as one event can start repo, scheduled and scoped workflows
|
||||
func (input *notifyInput) EncodedPayload() (string, error) {
|
||||
if input.encodedPayload == "" {
|
||||
p, err := json.Marshal(input.Payload)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("EncodedPayload: %w", err)
|
||||
}
|
||||
input.encodedPayload = string(p)
|
||||
}
|
||||
return input.encodedPayload, nil
|
||||
}
|
||||
|
||||
func (input *notifyInput) Notify(ctx context.Context) {
|
||||
log.Trace("execute %v for event %v whose doer is %v", getMethod(ctx), input.Event, input.Doer.Name)
|
||||
|
||||
@@ -337,16 +351,16 @@ func handleWorkflows(
|
||||
return nil
|
||||
}
|
||||
|
||||
p, err := json.Marshal(input.Payload)
|
||||
payload, err := input.EncodedPayload()
|
||||
if err != nil {
|
||||
return fmt.Errorf("json.Marshal: %w", err)
|
||||
return err
|
||||
}
|
||||
|
||||
isForkPullRequest := isForkPullRequestInput(input)
|
||||
|
||||
for _, dwf := range detectedWorkflows {
|
||||
// repo-level run: the workflow content is this repo at dwf.SourceCommitSHA
|
||||
if err := buildApproveAndInsertRun(ctx, input, ref, commit, string(p), isForkPullRequest, dwf, input.Repo.ID, false); err != nil {
|
||||
if err := buildApproveAndInsertRun(ctx, input, ref, commit, payload, isForkPullRequest, dwf, input.Repo.ID, false); err != nil {
|
||||
log.Error("repo %s: %v", input.Repo.FullName(), err)
|
||||
continue
|
||||
}
|
||||
@@ -595,9 +609,9 @@ func handleSchedules(
|
||||
return nil
|
||||
}
|
||||
|
||||
p, err := json.Marshal(input.Payload)
|
||||
payload, err := input.EncodedPayload()
|
||||
if err != nil {
|
||||
return fmt.Errorf("json.Marshal: %w", err)
|
||||
return err
|
||||
}
|
||||
|
||||
crons := make([]*actions_model.ActionSchedule, 0, len(detectedWorkflows))
|
||||
@@ -625,7 +639,7 @@ func handleSchedules(
|
||||
Ref: ref.String(),
|
||||
CommitSHA: commit.ID.String(),
|
||||
Event: input.Event,
|
||||
EventPayload: string(p),
|
||||
EventPayload: payload,
|
||||
Specs: schedules,
|
||||
Content: dwf.Content,
|
||||
}
|
||||
@@ -709,9 +723,9 @@ func detectAndHandleScopedWorkflows(
|
||||
return nil
|
||||
}
|
||||
|
||||
p, err := json.Marshal(input.Payload)
|
||||
payload, err := input.EncodedPayload()
|
||||
if err != nil {
|
||||
return fmt.Errorf("json.Marshal: %w", err)
|
||||
return err
|
||||
}
|
||||
isForkPullRequest := isForkPullRequestInput(input)
|
||||
actionsConfig := input.Repo.MustGetUnit(ctx, unit_model.TypeActions).ActionsConfig()
|
||||
@@ -759,7 +773,7 @@ func detectAndHandleScopedWorkflows(
|
||||
continue
|
||||
}
|
||||
|
||||
if err := buildApproveAndInsertRun(ctx, input, ref, consumerCommit, string(p), isForkPullRequest, dwf, sourceRepo.ID, true); err != nil {
|
||||
if err := buildApproveAndInsertRun(ctx, input, ref, consumerCommit, payload, isForkPullRequest, dwf, sourceRepo.ID, true); err != nil {
|
||||
log.Error("scoped workflows: source %s workflow %s: %v", sourceRepo.FullName(), dwf.EntryName, err)
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -46,6 +46,12 @@ func NotifyWorkflowJobsAndRunsStatusUpdate(ctx context.Context, jobs []*actions_
|
||||
}
|
||||
}
|
||||
|
||||
// notifyAndEmitFinishedJobs notifies status changes of jobs that may have finished and emits their runs' next jobs
|
||||
func notifyAndEmitFinishedJobs(ctx context.Context, jobs []*actions_model.ActionRunJob) {
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, jobs)
|
||||
EmitJobsIfReadyByJobs(jobs)
|
||||
}
|
||||
|
||||
// NotifyWorkflowRunStatusUpdateWithReload reloads the run before notifying its status update.
|
||||
// Use it when only repo/run IDs are available or when the in-memory run may be stale after job updates.
|
||||
func NotifyWorkflowRunStatusUpdateWithReload(ctx context.Context, repoID, runID int64) {
|
||||
|
||||
@@ -385,8 +385,7 @@ func execRerunPlan(ctx context.Context, plan *rerunPlan) (*actions_model.ActionR
|
||||
return nil, err
|
||||
}
|
||||
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, cancelledConcurrencyJobs)
|
||||
EmitJobsIfReadyByJobs(cancelledConcurrencyJobs)
|
||||
notifyAndEmitFinishedJobs(ctx, cancelledConcurrencyJobs)
|
||||
|
||||
CreateCommitStatusForRunJobs(ctx, plan.run, newJobs...)
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, newJobsToRerun)
|
||||
|
||||
@@ -169,8 +169,7 @@ func InsertRun(ctx context.Context, run *actions_model.ActionRun, content []byte
|
||||
return err
|
||||
}
|
||||
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, cancelledConcurrencyJobs)
|
||||
EmitJobsIfReadyByJobs(cancelledConcurrencyJobs)
|
||||
notifyAndEmitFinishedJobs(ctx, cancelledConcurrencyJobs)
|
||||
|
||||
// Post-commit kick: let the job emitter resolve jobs if needed
|
||||
if needPostCommitEmit {
|
||||
|
||||
@@ -720,7 +720,7 @@ func ResolveActionWorkflowForRun(ctx context.Context, repo *repo_model.Repositor
|
||||
}
|
||||
|
||||
// ToActionArtifact convert a actions_model.ActionArtifact to an api.ActionArtifact
|
||||
func ToActionArtifact(repo *repo_model.Repository, art *actions_model.ActionArtifact) (*api.ActionArtifact, error) {
|
||||
func ToActionArtifact(repo *repo_model.Repository, art *actions_model.ActionArtifact) *api.ActionArtifact {
|
||||
url := fmt.Sprintf("%s/actions/artifacts/%d", repo.APIURL(), art.ID)
|
||||
|
||||
return &api.ActionArtifact{
|
||||
@@ -738,10 +738,10 @@ func ToActionArtifact(repo *repo_model.Repository, art *actions_model.ActionArti
|
||||
RepositoryID: art.RepoID,
|
||||
HeadSha: art.CommitSHA,
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
|
||||
func ToActionRunner(ctx context.Context, runner *actions_model.ActionRunner) *api.ActionRunner {
|
||||
func ToActionRunner(runner *actions_model.ActionRunner) *api.ActionRunner {
|
||||
status := runner.Status()
|
||||
apiStatus := "offline"
|
||||
if runner.IsOnline() {
|
||||
|
||||
Reference in new issue
Block a user