mirror of
https://github.com/go-gitea/gitea.git
synced 2026-11-09 02:13:47 +00:00
Backport #40126 by @silverwind The job emitter resolves a dependency chain in several rounds in memory, but a job's `if:` reads its needs' results from the database, where a need skipped or failed earlier in the same pass was still blocked. A job with `if: always() && needs.build.result == 'skipped'` was therefore skipped instead of started. Finished statuses are now persisted at the end of each round, which also replaces the separate handling of skipped and failed reusable workflow callers. Co-authored-by: silverwind <me@silverwind.io>
This commit is contained in:
1 parent
5aea433c71
commit
db5df0494a
2 files changed
+53
-38
No files matched your search
@@ -309,50 +309,41 @@ func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.Action
|
||||
continue
|
||||
}
|
||||
|
||||
if status.IsDone() {
|
||||
result.UpdatedJobs = append(result.UpdatedJobs, job)
|
||||
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 {
|
||||
if err := upsertJobErrorSummary(ctx, job, "uses", err); err != nil {
|
||||
return err
|
||||
}
|
||||
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, actions_model.StatusFailure:
|
||||
job.Status = status
|
||||
if n, err := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": actions_model.StatusBlocked}, "status"); err != nil {
|
||||
return err
|
||||
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 {
|
||||
if err := upsertJobErrorSummary(ctx, job, "uses", err); err != nil {
|
||||
return err
|
||||
}
|
||||
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
|
||||
}
|
||||
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 {
|
||||
if err := updateBlockedJobStatus(ctx, job, 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)
|
||||
}
|
||||
@@ -497,9 +488,14 @@ func (r *jobStatusResolver) Resolve(ctx context.Context) (map[int64]actions_mode
|
||||
if len(updated) == 0 {
|
||||
return ret, nil
|
||||
}
|
||||
for k, v := range updated {
|
||||
ret[k] = v
|
||||
r.statuses[k] = v
|
||||
for id, status := range updated {
|
||||
if status.IsDone() { // later rounds evaluate `if:` against needs read from the database
|
||||
if err := updateBlockedJobStatus(ctx, r.jobMap[id], status); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
ret[id] = status
|
||||
r.statuses[id] = status
|
||||
}
|
||||
if r.matrixInserted {
|
||||
// Matrix expansion inserted sibling rows this round. They are not in statuses/needs, so
|
||||
@@ -532,6 +528,16 @@ func (r *jobStatusResolver) resolveCheckNeeds(id int64) (allDone, allSucceed boo
|
||||
return allDone, allSucceed
|
||||
}
|
||||
|
||||
func updateBlockedJobStatus(ctx context.Context, job *actions_model.ActionRunJob, status actions_model.Status) error {
|
||||
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)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *jobStatusResolver) updateStatus(ctx context.Context, job *actions_model.ActionRunJob, status actions_model.Status) error {
|
||||
cond := builder.Eq{"status": job.Status}
|
||||
job.Status = status
|
||||
|
||||
@@ -91,6 +91,15 @@ func Test_jobStatusResolver_Resolve(t *testing.T) {
|
||||
},
|
||||
want: map[int64]actions_model.Status{3: actions_model.StatusWaiting},
|
||||
},
|
||||
{
|
||||
name: "`if` reads the result of a need finished in an earlier round",
|
||||
jobs: actions_model.ActionJobList{
|
||||
{ID: 1, JobID: "a", Status: actions_model.StatusFailure},
|
||||
{ID: 2, JobID: "b", Status: actions_model.StatusBlocked, Needs: []string{"a"}},
|
||||
{ID: 3, JobID: "c", Status: actions_model.StatusBlocked, Needs: []string{"b"}, WorkflowPayload: []byte(`jobs: {c: {if: "${{ always() && needs.b.result == 'skipped' }}"}}`)},
|
||||
},
|
||||
want: map[int64]actions_model.Status{2: actions_model.StatusSkipped, 3: actions_model.StatusWaiting},
|
||||
},
|
||||
{
|
||||
name: "loop need",
|
||||
jobs: actions_model.ActionJobList{
|
||||
|
||||
Reference in new issue
Block a user