This commit is contained in:
@@ -8,15 +8,15 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
actions_model "code.gitea.io/gitea/models/actions"
|
||||
"code.gitea.io/gitea/models/db"
|
||||
"code.gitea.io/gitea/modules/container"
|
||||
"code.gitea.io/gitea/modules/graceful"
|
||||
"code.gitea.io/gitea/modules/log"
|
||||
"code.gitea.io/gitea/modules/queue"
|
||||
"code.gitea.io/gitea/modules/setting"
|
||||
"code.gitea.io/gitea/modules/util"
|
||||
notify_service "code.gitea.io/gitea/services/notify"
|
||||
actions_model "gitea.dev/models/actions"
|
||||
"gitea.dev/models/db"
|
||||
"gitea.dev/modules/container"
|
||||
"gitea.dev/modules/graceful"
|
||||
"gitea.dev/modules/log"
|
||||
"gitea.dev/modules/queue"
|
||||
"gitea.dev/modules/setting"
|
||||
"gitea.dev/modules/timeutil"
|
||||
"gitea.dev/modules/util"
|
||||
|
||||
"xorm.io/builder"
|
||||
)
|
||||
@@ -27,7 +27,7 @@ type jobUpdate struct {
|
||||
RunID int64
|
||||
}
|
||||
|
||||
func EmitJobsIfReadyByRun(runID int64) error {
|
||||
var EmitJobsIfReadyByRun = func(runID int64) error {
|
||||
err := jobEmitterQueue.Push(&jobUpdate{
|
||||
RunID: runID,
|
||||
})
|
||||
@@ -70,36 +70,48 @@ func checkJobsByRunID(ctx context.Context, runID int64) error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("get action run: %w", err)
|
||||
}
|
||||
var jobs, updatedJobs []*actions_model.ActionRunJob
|
||||
var result jobsCheckResult
|
||||
if err := db.WithTx(ctx, func(ctx context.Context) error {
|
||||
// check jobs of the current run
|
||||
if js, ujs, err := checkJobsOfRun(ctx, run); err != nil {
|
||||
r, err := checkJobsOfCurrentRunAttempt(ctx, run)
|
||||
if err != nil {
|
||||
return err
|
||||
} else {
|
||||
jobs = append(jobs, js...)
|
||||
updatedJobs = append(updatedJobs, ujs...)
|
||||
}
|
||||
if js, ujs, err := checkRunConcurrency(ctx, run); err != nil {
|
||||
result.merge(r)
|
||||
|
||||
r, err = checkRunConcurrency(ctx, run)
|
||||
if err != nil {
|
||||
return err
|
||||
} else {
|
||||
jobs = append(jobs, js...)
|
||||
updatedJobs = append(updatedJobs, ujs...)
|
||||
}
|
||||
result.merge(r)
|
||||
return nil
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
CreateCommitStatusForRunJobs(ctx, run, jobs...)
|
||||
for _, job := range updatedJobs {
|
||||
_ = job.LoadAttributes(ctx)
|
||||
notify_service.WorkflowJobStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job, nil)
|
||||
// Re-emit AFTER the transaction commits; doing this inside WithTx would deadlock under
|
||||
// immediate-mode queues (the inline handler reopens checkJobsByRunID and asks for a
|
||||
// nested writer transaction while the outer one is still open).
|
||||
emitted := make(container.Set[int64])
|
||||
for _, rid := range result.RunIDsToReEmit {
|
||||
if !emitted.Add(rid) {
|
||||
continue
|
||||
}
|
||||
if err := EmitJobsIfReadyByRun(rid); err != nil {
|
||||
log.Error("re-emit run %d: %v", rid, err)
|
||||
}
|
||||
}
|
||||
NotifyWorkflowJobsAndRunsStatusUpdate(ctx, result.CancelledJobs)
|
||||
EmitJobsIfReadyByJobs(result.CancelledJobs)
|
||||
if err := createCommitStatusesForJobsByRun(ctx, result.Jobs); err != nil {
|
||||
return err
|
||||
}
|
||||
NotifyWorkflowJobsStatusUpdate(ctx, result.UpdatedJobs...)
|
||||
runJobs := make(map[int64][]*actions_model.ActionRunJob)
|
||||
for _, job := range jobs {
|
||||
for _, job := range result.Jobs {
|
||||
runJobs[job.RunID] = append(runJobs[job.RunID], job)
|
||||
}
|
||||
runUpdatedJobs := make(map[int64][]*actions_model.ActionRunJob)
|
||||
for _, uj := range updatedJobs {
|
||||
for _, uj := range result.UpdatedJobs {
|
||||
runUpdatedJobs[uj.RunID] = append(runUpdatedJobs[uj.RunID], uj)
|
||||
}
|
||||
for runID, js := range runJobs {
|
||||
@@ -114,71 +126,106 @@ func checkJobsByRunID(ctx context.Context, runID int64) error {
|
||||
}
|
||||
}
|
||||
if runUpdated {
|
||||
NotifyWorkflowRunStatusUpdateWithReload(ctx, js[0])
|
||||
NotifyWorkflowRunStatusUpdateWithReload(ctx, js[0].RepoID, js[0].RunID)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// findBlockedRunByConcurrency finds the blocked concurrent run in a repo and returns `nil, nil` when there is no blocked run.
|
||||
func findBlockedRunByConcurrency(ctx context.Context, repoID int64, concurrencyGroup string) (*actions_model.ActionRun, error) {
|
||||
if concurrencyGroup == "" {
|
||||
return nil, nil //nolint:nilnil // return nil to indicate that no blocked run exists
|
||||
}
|
||||
cRuns, cJobs, err := actions_model.GetConcurrentRunsAndJobs(ctx, repoID, concurrencyGroup, []actions_model.Status{actions_model.StatusBlocked})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("find concurrent runs and jobs: %w", err)
|
||||
func createCommitStatusesForJobsByRun(ctx context.Context, jobs []*actions_model.ActionRunJob) error {
|
||||
runJobs := make(map[int64][]*actions_model.ActionRunJob)
|
||||
for _, job := range jobs {
|
||||
runJobs[job.RunID] = append(runJobs[job.RunID], job)
|
||||
}
|
||||
|
||||
// There can be at most one blocked run or job
|
||||
var concurrentRun *actions_model.ActionRun
|
||||
if len(cRuns) > 0 {
|
||||
concurrentRun = cRuns[0]
|
||||
} else if len(cJobs) > 0 {
|
||||
jobRun, exist, err := db.GetByID[actions_model.ActionRun](ctx, cJobs[0].RunID)
|
||||
if !exist {
|
||||
return nil, fmt.Errorf("run %d does not exist", cJobs[0].RunID)
|
||||
}
|
||||
for jobRunID, jobList := range runJobs {
|
||||
run, err := actions_model.GetRunByRepoAndID(ctx, jobList[0].RepoID, jobRunID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("get run by job %d: %w", cJobs[0].ID, err)
|
||||
return fmt.Errorf("get action run %d: %w", jobRunID, err)
|
||||
}
|
||||
concurrentRun = jobRun
|
||||
CreateCommitStatusForRunJobs(ctx, run, jobList...)
|
||||
}
|
||||
|
||||
return concurrentRun, nil
|
||||
return nil
|
||||
}
|
||||
|
||||
func checkRunConcurrency(ctx context.Context, run *actions_model.ActionRun) (jobs, updatedJobs []*actions_model.ActionRunJob, err error) {
|
||||
// findConcurrencyWaiterToWake returns a run (other than excludeRunID) blocked on the group that can be woken now
|
||||
func findConcurrencyWaiterToWake(ctx context.Context, repoID, excludeRunID int64, concurrencyGroup string) (int64, error) {
|
||||
if concurrencyGroup == "" {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// The slot should be free before any waiter can proceed.
|
||||
holderAttempts, holderJobs, err := actions_model.GetConcurrentRunAttemptsAndJobs(ctx, repoID, concurrencyGroup, []actions_model.Status{actions_model.StatusRunning, actions_model.StatusCancelling})
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("find concurrency-group holders: %w", err)
|
||||
}
|
||||
if len(holderAttempts) > 0 || len(holderJobs) > 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
cAttempts, cJobs, err := actions_model.GetConcurrentRunAttemptsAndJobs(ctx, repoID, concurrencyGroup, []actions_model.Status{actions_model.StatusBlocked})
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("find blocked concurrent runs: %w", err)
|
||||
}
|
||||
for _, a := range cAttempts {
|
||||
if a.RunID != excludeRunID {
|
||||
return a.RunID, nil
|
||||
}
|
||||
}
|
||||
for _, j := range cJobs {
|
||||
if j.RunID != excludeRunID {
|
||||
return j.RunID, nil
|
||||
}
|
||||
}
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
// checkRunConcurrency wakes a run blocked by concurrency that may become runnable now that
|
||||
// the current run's activity may have freed a workflow-level or job-level concurrency group.
|
||||
func checkRunConcurrency(ctx context.Context, run *actions_model.ActionRun) (*jobsCheckResult, error) {
|
||||
result := &jobsCheckResult{}
|
||||
checkedConcurrencyGroup := make(container.Set[string])
|
||||
|
||||
collect := func(concurrencyGroup string) error {
|
||||
concurrentRun, err := findBlockedRunByConcurrency(ctx, run.RepoID, concurrencyGroup)
|
||||
if err != nil {
|
||||
return fmt.Errorf("find blocked run by concurrency: %w", err)
|
||||
}
|
||||
if concurrentRun != nil && !concurrentRun.NeedApproval {
|
||||
js, ujs, err := checkJobsOfRun(ctx, concurrentRun)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
jobs = append(jobs, js...)
|
||||
updatedJobs = append(updatedJobs, ujs...)
|
||||
}
|
||||
checkedConcurrencyGroup.Add(concurrencyGroup)
|
||||
|
||||
// Exclude run.ID: this run's own jobs are resolved by checkJobsOfCurrentRunAttempt, no need to re-emit.
|
||||
concurrentRunID, err := findConcurrencyWaiterToWake(ctx, run.RepoID, run.ID, concurrencyGroup)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if concurrentRunID == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
concurrentRun, err := actions_model.GetRunByRepoAndID(ctx, run.RepoID, concurrentRunID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("get concurrent run %d: %w", concurrentRunID, err)
|
||||
}
|
||||
// A run awaiting approval is not advanced by concurrency; ApproveRuns emits it once approved.
|
||||
if concurrentRun.NeedApproval {
|
||||
return nil
|
||||
}
|
||||
|
||||
result.RunIDsToReEmit = append(result.RunIDsToReEmit, concurrentRunID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// check run (workflow-level) concurrency
|
||||
if run.ConcurrencyGroup != "" {
|
||||
if err := collect(run.ConcurrencyGroup); err != nil {
|
||||
return nil, nil, err
|
||||
runConcurrencyGroup, _, err := run.GetEffectiveConcurrency(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("GetEffectiveConcurrency: %w", err)
|
||||
}
|
||||
if runConcurrencyGroup != "" {
|
||||
if err := collect(runConcurrencyGroup); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// check job concurrency
|
||||
runJobs, err := db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: run.ID})
|
||||
runJobs, err := actions_model.GetLatestAttemptJobsByRepoAndRunID(ctx, run.RepoID, run.ID)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("find run %d jobs: %w", run.ID, err)
|
||||
return nil, fmt.Errorf("find run %d jobs: %w", run.ID, err)
|
||||
}
|
||||
for _, job := range runJobs {
|
||||
if !job.Status.IsDone() {
|
||||
@@ -188,83 +235,138 @@ func checkRunConcurrency(ctx context.Context, run *actions_model.ActionRun) (job
|
||||
continue
|
||||
}
|
||||
if err := collect(job.ConcurrencyGroup); err != nil {
|
||||
return nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return jobs, updatedJobs, nil
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func checkJobsOfRun(ctx context.Context, run *actions_model.ActionRun) (jobs, updatedJobs []*actions_model.ActionRunJob, err error) {
|
||||
jobs, err = db.Find[actions_model.ActionRunJob](ctx, actions_model.FindRunJobOptions{RunID: run.ID})
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
// checkJobsOfCurrentRunAttempt resolves blocked jobs of the run's latest attempt.
|
||||
func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.ActionRun) (*jobsCheckResult, error) {
|
||||
// Approval is the only transition allowed to release an approval-pending run.
|
||||
if run.NeedApproval {
|
||||
return &jobsCheckResult{}, nil
|
||||
}
|
||||
|
||||
jobs, err := actions_model.GetRunJobsByRunAndAttemptID(ctx, run.ID, run.LatestAttemptID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
result := &jobsCheckResult{Jobs: jobs}
|
||||
|
||||
var attempt *actions_model.ActionRunAttempt
|
||||
if run.LatestAttemptID > 0 {
|
||||
attempt, err = actions_model.GetRunAttemptByRepoAndID(ctx, run.RepoID, run.LatestAttemptID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// The resolver below only considers needs and job-level concurrency, so a run blocked
|
||||
// solely by run-level concurrency would have its jobs unblocked here. checkRunConcurrency
|
||||
// re-evaluates when the holding run finishes.
|
||||
if run.Status.IsBlocked() {
|
||||
shouldBlock, err := shouldBlockRunByConcurrency(ctx, run)
|
||||
if run.Status.IsBlocked() && attempt != nil {
|
||||
shouldBlock, err := shouldBlockRunByConcurrency(ctx, attempt)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("shouldBlockRunByConcurrency: %w", err)
|
||||
return nil, fmt.Errorf("shouldBlockRunByConcurrency: %w", err)
|
||||
}
|
||||
if shouldBlock {
|
||||
return jobs, nil, nil
|
||||
return result, nil
|
||||
}
|
||||
}
|
||||
vars, err := actions_model.GetVariablesOfRun(ctx, run)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
resolver := newJobStatusResolver(jobs, vars)
|
||||
|
||||
expandedAnyCaller := false
|
||||
if err = db.WithTx(ctx, func(ctx context.Context) error {
|
||||
for _, job := range jobs {
|
||||
job.Run = run
|
||||
}
|
||||
|
||||
updates := newJobStatusResolver(jobs, vars).Resolve(ctx)
|
||||
updates := resolver.Resolve(ctx)
|
||||
for _, job := range jobs {
|
||||
if status, ok := updates[job.ID]; ok {
|
||||
job.Status = status
|
||||
if n, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": actions_model.StatusBlocked}, "status"); err != nil {
|
||||
return err
|
||||
} else if n != 1 {
|
||||
return fmt.Errorf("no affected for updating blocked job %v", job.ID)
|
||||
}
|
||||
updatedJobs = append(updatedJobs, job)
|
||||
status, ok := updates[job.ID]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
if job.IsReusableCaller {
|
||||
switch status {
|
||||
case actions_model.StatusWaiting:
|
||||
if err := expandReusableWorkflowCaller(ctx, run, attempt, job, vars); err != nil {
|
||||
// Terminal expansion failure (an invalid/unresolvable reusable workflow): fail this caller.
|
||||
log.Warn("caller %d cannot be expanded: %v", job.ID, err)
|
||||
job.Status = actions_model.StatusFailure
|
||||
job.Stopped = timeutil.TimeStampNow()
|
||||
if n, uerr := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": actions_model.StatusBlocked, "is_expanded": false}, "status", "stopped"); uerr != nil {
|
||||
return fmt.Errorf("mark unexpandable caller %d failed: %w", job.ID, uerr)
|
||||
} else if n == 1 {
|
||||
log.Warn("unexpandable caller %d has been marked as failed", job.ID)
|
||||
result.UpdatedJobs = append(result.UpdatedJobs, job)
|
||||
// Re-emit so the failed caller's dependents get resolved on the next pass.
|
||||
expandedAnyCaller = true
|
||||
} else {
|
||||
// A concurrent writer advanced the caller; restore the in-memory state.
|
||||
log.Warn("unexpandable caller %d has been advanced by a concurrent writer, not marking it failed", job.ID)
|
||||
job.Status = actions_model.StatusBlocked
|
||||
job.Stopped = 0
|
||||
}
|
||||
} else {
|
||||
expandedAnyCaller = true
|
||||
}
|
||||
case actions_model.StatusSkipped:
|
||||
job.Status = actions_model.StatusSkipped
|
||||
if _, err := actions_model.UpdateRunJob(ctx, job, nil, "status"); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Non-caller: standard status update.
|
||||
job.Status = status
|
||||
if n, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": actions_model.StatusBlocked}, "status"); err != nil {
|
||||
return err
|
||||
} else if n != 1 {
|
||||
return fmt.Errorf("no affected for updating blocked job %v", job.ID)
|
||||
}
|
||||
result.UpdatedJobs = append(result.UpdatedJobs, job)
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return jobs, updatedJobs, nil
|
||||
}
|
||||
|
||||
func NotifyWorkflowRunStatusUpdateWithReload(ctx context.Context, job *actions_model.ActionRunJob) {
|
||||
job.Run = nil
|
||||
if err := job.LoadAttributes(ctx); err != nil {
|
||||
log.Error("LoadAttributes: %v", err)
|
||||
return
|
||||
if expandedAnyCaller {
|
||||
result.RunIDsToReEmit = append(result.RunIDsToReEmit, run.ID)
|
||||
}
|
||||
notify_service.WorkflowRunStatusUpdate(ctx, job.Run.Repo, job.Run.TriggerUser, job.Run)
|
||||
|
||||
// Recomputes the repository's num_action_runs / num_closed_action_runs counters since the run's status changed
|
||||
actions_model.UpdateRepoRunsNumbers(ctx, job.RepoID)
|
||||
result.CancelledJobs = resolver.cancelledJobs
|
||||
return result, nil
|
||||
}
|
||||
|
||||
type jobStatusResolver struct {
|
||||
statuses map[int64]actions_model.Status
|
||||
needs map[int64][]int64
|
||||
jobMap map[int64]*actions_model.ActionRunJob
|
||||
vars map[string]string
|
||||
statuses map[int64]actions_model.Status
|
||||
needs map[int64][]int64
|
||||
jobMap map[int64]*actions_model.ActionRunJob
|
||||
vars map[string]string
|
||||
cancelledJobs []*actions_model.ActionRunJob
|
||||
}
|
||||
|
||||
func newJobStatusResolver(jobs actions_model.ActionJobList, vars map[string]string) *jobStatusResolver {
|
||||
idToJobs := make(map[string][]*actions_model.ActionRunJob, len(jobs))
|
||||
// Scope-aware: needs are resolved within the same ParentJobID scope so the same
|
||||
// JobID in different reusable workflow calls does not cross-link.
|
||||
scopedIDToJobs := make(map[int64]map[string][]*actions_model.ActionRunJob)
|
||||
jobMap := make(map[int64]*actions_model.ActionRunJob)
|
||||
for _, job := range jobs {
|
||||
idToJobs[job.JobID] = append(idToJobs[job.JobID], job)
|
||||
scope := scopedIDToJobs[job.ParentJobID]
|
||||
if scope == nil {
|
||||
scope = make(map[string][]*actions_model.ActionRunJob)
|
||||
scopedIDToJobs[job.ParentJobID] = scope
|
||||
}
|
||||
scope[job.JobID] = append(scope[job.JobID], job)
|
||||
jobMap[job.ID] = job
|
||||
}
|
||||
|
||||
@@ -272,8 +374,9 @@ func newJobStatusResolver(jobs actions_model.ActionJobList, vars map[string]stri
|
||||
needs := make(map[int64][]int64, len(jobs))
|
||||
for _, job := range jobs {
|
||||
statuses[job.ID] = job.Status
|
||||
scope := scopedIDToJobs[job.ParentJobID]
|
||||
for _, need := range job.Needs {
|
||||
for _, v := range idToJobs[need] {
|
||||
for _, v := range scope[need] {
|
||||
needs[job.ID] = append(needs[job.ID], v.ID)
|
||||
}
|
||||
}
|
||||
@@ -308,6 +411,11 @@ func (r *jobStatusResolver) resolveCheckNeeds(id int64) (allDone, allSucceed boo
|
||||
if !needStatus.IsDone() {
|
||||
allDone = false
|
||||
}
|
||||
// A failed need with continue-on-error:true is treated as success, matching AggregateJobStatus,
|
||||
// so a downstream job with an implicit `success()` is not skipped.
|
||||
if needJob := r.jobMap[need]; needJob != nil && needJob.ContinueOnError && needStatus == actions_model.StatusFailure {
|
||||
continue
|
||||
}
|
||||
if needStatus.In(actions_model.StatusFailure, actions_model.StatusCancelled, actions_model.StatusSkipped) {
|
||||
allSucceed = false
|
||||
}
|
||||
@@ -315,14 +423,6 @@ func (r *jobStatusResolver) resolveCheckNeeds(id int64) (allDone, allSucceed boo
|
||||
return allDone, allSucceed
|
||||
}
|
||||
|
||||
func (r *jobStatusResolver) resolveJobHasIfCondition(actionRunJob *actions_model.ActionRunJob) (hasIf bool) {
|
||||
// FIXME evaluate this on the server side
|
||||
if job, err := actionRunJob.ParseJob(); err == nil {
|
||||
return len(job.If.Value) > 0
|
||||
}
|
||||
return hasIf
|
||||
}
|
||||
|
||||
func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model.Status {
|
||||
ret := map[int64]actions_model.Status{}
|
||||
for id, status := range r.statuses {
|
||||
@@ -330,6 +430,12 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model
|
||||
if status != actions_model.StatusBlocked {
|
||||
continue
|
||||
}
|
||||
// A child of a caller cannot start until the caller has become "ready" (children inserted, CallPayload populated).
|
||||
if actionRunJob.ParentJobID > 0 {
|
||||
if parent, ok := r.jobMap[actionRunJob.ParentJobID]; ok && !parent.IsExpanded {
|
||||
continue
|
||||
}
|
||||
}
|
||||
allDone, allSucceed := r.resolveCheckNeeds(id)
|
||||
if !allDone {
|
||||
continue
|
||||
@@ -340,25 +446,26 @@ func (r *jobStatusResolver) resolve(ctx context.Context) map[int64]actions_model
|
||||
if err != nil {
|
||||
// The err can be caused by different cases: database error, or syntax error, or the needed jobs haven't completed
|
||||
// At the moment there is no way to distinguish them.
|
||||
// Actually, for most cases, the error is caused by "syntax error" / "the needed jobs haven't completed (skipped?)"
|
||||
// TODO: if workflow or concurrency expression has syntax error, there should be a user error message, need to show it to end users
|
||||
log.Debug("updateConcurrencyEvaluationForJobWithNeeds failed, this job will stay blocked: job: %d, err: %v", id, err)
|
||||
continue
|
||||
}
|
||||
|
||||
shouldStartJob := true
|
||||
if !allSucceed {
|
||||
// Not all dependent jobs completed successfully:
|
||||
// * if the job has "if" condition, it can be started, then the act_runner will evaluate the "if" condition.
|
||||
// * otherwise, the job should be skipped.
|
||||
shouldStartJob = r.resolveJobHasIfCondition(actionRunJob)
|
||||
shouldStartJob, err := evaluateJobIf(ctx, actionRunJob.Run, nil, actionRunJob, r.vars, allSucceed)
|
||||
if err != nil {
|
||||
// TODO: surface deterministic expression errors to users by failing the job with a message.
|
||||
log.Error("evaluateJobIf failed, job will stay blocked: job: %d, err: %v", id, err)
|
||||
continue
|
||||
}
|
||||
|
||||
newStatus := util.Iif(shouldStartJob, actions_model.StatusWaiting, actions_model.StatusSkipped)
|
||||
if newStatus == actions_model.StatusWaiting {
|
||||
newStatus, err = PrepareToStartJobWithConcurrency(ctx, actionRunJob)
|
||||
var cancelledJobs []*actions_model.ActionRunJob
|
||||
newStatus, cancelledJobs, err = PrepareToStartJobWithConcurrency(ctx, actionRunJob)
|
||||
if err != nil {
|
||||
log.Error("ShouldBlockJobByConcurrency failed, this job will stay blocked: job: %d, err: %v", id, err)
|
||||
} else {
|
||||
r.cancelledJobs = append(r.cancelledJobs, cancelledJobs...)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -374,8 +481,16 @@ func updateConcurrencyEvaluationForJobWithNeeds(ctx context.Context, actionRunJo
|
||||
return nil // for testing purpose only, no repo, no evaluation
|
||||
}
|
||||
|
||||
err := EvaluateJobConcurrencyFillModel(ctx, actionRunJob.Run, actionRunJob, vars, nil)
|
||||
if err != nil {
|
||||
// Legacy jobs (created before migration v331) have RunAttemptID=0 and no attempt record.
|
||||
var attempt *actions_model.ActionRunAttempt
|
||||
if actionRunJob.RunAttemptID > 0 {
|
||||
var err error
|
||||
attempt, err = actions_model.GetRunAttemptByRepoAndID(ctx, actionRunJob.RepoID, actionRunJob.RunAttemptID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("GetRunAttemptByRepoAndID: %w", err)
|
||||
}
|
||||
}
|
||||
if err := EvaluateJobConcurrencyFillModel(ctx, actionRunJob.Run, attempt, actionRunJob, vars, nil); err != nil {
|
||||
return fmt.Errorf("evaluate job concurrency: %w", err)
|
||||
}
|
||||
|
||||
@@ -384,3 +499,23 @@ func updateConcurrencyEvaluationForJobWithNeeds(ctx context.Context, actionRunJo
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// jobsCheckResult bundles the output of the per-run job-check helpers.
|
||||
type jobsCheckResult struct {
|
||||
// Jobs are all jobs of the run's latest attempt that were inspected.
|
||||
Jobs []*actions_model.ActionRunJob
|
||||
// UpdatedJobs are jobs whose status was transitioned out of Blocked in this pass.
|
||||
UpdatedJobs []*actions_model.ActionRunJob
|
||||
// CancelledJobs are jobs cancelled by job-level concurrency while preparing to start.
|
||||
CancelledJobs []*actions_model.ActionRunJob
|
||||
// RunIDsToReEmit are runs that need another resolver pass in their own transaction.
|
||||
RunIDsToReEmit []int64
|
||||
}
|
||||
|
||||
// merge appends another result's contents into r in place.
|
||||
func (r *jobsCheckResult) merge(other *jobsCheckResult) {
|
||||
r.Jobs = append(r.Jobs, other.Jobs...)
|
||||
r.UpdatedJobs = append(r.UpdatedJobs, other.UpdatedJobs...)
|
||||
r.CancelledJobs = append(r.CancelledJobs, other.CancelledJobs...)
|
||||
r.RunIDsToReEmit = append(r.RunIDsToReEmit, other.RunIDsToReEmit...)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user