From db5df0494a02c4ea7b822b33c4a305918258d7fe Mon Sep 17 00:00:00 2001 From: Giteabot Date: Wed, 4 Nov 2026 11:17:17 -0800 Subject: [PATCH] fix(actions): evaluate job `if:` against needs finished in the same pass (#40126) (#40221) 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 --- services/actions/job_emitter.go | 82 +++++++++++++++------------- services/actions/job_emitter_test.go | 9 +++ 2 files changed, 53 insertions(+), 38 deletions(-) diff --git a/services/actions/job_emitter.go b/services/actions/job_emitter.go index 114e9d75ab3..69b111e2340 100644 --- a/services/actions/job_emitter.go +++ b/services/actions/job_emitter.go @@ -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 diff --git a/services/actions/job_emitter_test.go b/services/actions/job_emitter_test.go index 87702937cc5..115ab3fcf56 100644 --- a/services/actions/job_emitter_test.go +++ b/services/actions/job_emitter_test.go @@ -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{