diff --git a/models/actions/run_job.go b/models/actions/run_job.go index e2f6e7c992d..5f7a5367e17 100644 --- a/models/actions/run_job.go +++ b/models/actions/run_job.go @@ -524,58 +524,74 @@ func UpdateRunJob(ctx context.Context, job *ActionRunJob, cond builder.Cond, col return affected, RefreshReusableCallerStatus(ctx, parent) } - { - // Other goroutines may aggregate the status of the attempt/run and update it too. - // So we need to load the current jobs before updating the aggregate state. - if job.RunAttemptID > 0 { - attempt, err := GetRunAttemptByRepoAndID(ctx, job.RepoID, job.RunAttemptID) - if err != nil { - return 0, err - } - jobs, err := GetRunJobsByRunAndAttemptID(ctx, job.RunID, job.RunAttemptID) - if err != nil { - return 0, err - } - attempt.Status = AggregateJobStatus(jobs) - if attempt.Started.IsZero() && attempt.Status.IsRunning() { - attempt.Started = timeutil.TimeStampNow() - } - if attempt.Stopped.IsZero() && attempt.Status.IsDone() { - attempt.Stopped = timeutil.TimeStampNow() - } - if err := UpdateRunAttempt(ctx, attempt, "status", "started", "stopped"); err != nil { - return 0, fmt.Errorf("update run attempt %d: %w", attempt.ID, err) - } - } else { - // TODO: Remove this fallback in the future. - // Legacy fallback: jobs created before migration v331 have RunAttemptID=0 and are NOT backfilled. - // This path keeps those runs' status consistent when their jobs finish, including: - // - jobs created before migration v331 and complete on the new version starts - // - zombie/abandoned cleanup cron tasks that call UpdateRunJob on legacy jobs - run, err := GetRunByRepoAndID(ctx, job.RepoID, job.RunID) - if err != nil { - return 0, err - } - jobs, err := GetLatestAttemptJobsByRepoAndRunID(ctx, job.RepoID, job.RunID) - if err != nil { - return 0, err - } - run.Status = AggregateJobStatus(jobs) - if run.Started.IsZero() && run.Status.IsRunning() { - run.Started = timeutil.TimeStampNow() - } - if run.Stopped.IsZero() && run.Status.IsDone() { - run.Stopped = timeutil.TimeStampNow() - } - if err := UpdateRun(ctx, run, "status", "started", "stopped"); err != nil { - return 0, fmt.Errorf("update run %d: %w", run.ID, err) - } - } + if err := refreshRunStatus(ctx, job.RepoID, job.RunID, job.RunAttemptID, StatusUnknown); err != nil { + return 0, err } return affected, nil } +// refreshRunStatus recomputes the status of an attempt from the jobs currently stored and persists it. +// The latest attempt propagates its status to its run, an older one only updates itself. +// noJobsStatus settles an attempt without any job, which AggregateJobStatus cannot conclude on its own. +func refreshRunStatus(ctx context.Context, repoID, runID, runAttemptID int64, noJobsStatus Status) error { + // Other goroutines may aggregate the status of the attempt/run and update it too. + // So we need to load the current jobs before updating the aggregate state. + if runAttemptID > 0 { + attempt, err := GetRunAttemptByRepoAndID(ctx, repoID, runAttemptID) + if err != nil { + return err + } + jobs, err := GetRunJobsByRunAndAttemptID(ctx, runID, runAttemptID) + if err != nil { + return err + } + attempt.Status = AggregateJobStatus(jobs) + if len(jobs) == 0 { + attempt.Status = noJobsStatus + } + if attempt.Started.IsZero() && attempt.Status.IsRunning() { + attempt.Started = timeutil.TimeStampNow() + } + if attempt.Stopped.IsZero() && attempt.Status.IsDone() { + attempt.Stopped = timeutil.TimeStampNow() + } + if err := UpdateRunAttempt(ctx, attempt, "status", "started", "stopped"); err != nil { + return fmt.Errorf("update run attempt %d: %w", attempt.ID, err) + } + return nil + } + + // TODO: Remove this fallback in the future. + // Legacy fallback: jobs created before migration v331 have RunAttemptID=0 and are NOT backfilled. + // This path keeps those runs' status consistent when their jobs finish, including: + // - jobs created before migration v331 and complete on the new version starts + // - zombie/abandoned cleanup cron tasks that call UpdateRunJob on legacy jobs + // - cancelling a legacy run whose jobs are all already done + run, err := GetRunByRepoAndID(ctx, repoID, runID) + if err != nil { + return err + } + jobs, err := GetLatestAttemptJobsByRun(ctx, run) + if err != nil { + return err + } + run.Status = AggregateJobStatus(jobs) + if len(jobs) == 0 { + run.Status = noJobsStatus + } + if run.Started.IsZero() && run.Status.IsRunning() { + run.Started = timeutil.TimeStampNow() + } + if run.Stopped.IsZero() && run.Status.IsDone() { + run.Stopped = timeutil.TimeStampNow() + } + if err := UpdateRun(ctx, run, "status", "started", "stopped"); err != nil { + return fmt.Errorf("update run %d: %w", run.ID, err) + } + return nil +} + // RefreshReusableCallerStatus recomputes a reusable workflow caller's Status, Started and Stopped from its current direct children and persists the change. // No-op if caller is not a reusable caller. // @@ -736,6 +752,8 @@ func CancelPreviousJobsByJobConcurrency(ctx context.Context, job *ActionRunJob) return CancelJobs(ctx, jobsToCancel) } +// CancelJobs cancels every cancellable job it is given. It leaves the status of a run it +// cancelled nothing in untouched, SettleRunAfterCancel is what gives such a run a final one. func CancelJobs(ctx context.Context, jobs []*ActionRunJob) ([]*ActionRunJob, error) { cancelledJobs := make([]*ActionRunJob, 0, len(jobs)) @@ -760,6 +778,16 @@ func CancelJobs(ctx context.Context, jobs []*ActionRunJob) ([]*ActionRunJob, err return cancelledJobs, nil } +// SettleRunAfterCancel gives a run a final status when cancelling it updated no job at all. +// A run's status is otherwise only ever written as a side effect of a job update, so a run whose +// jobs are all done already, or that has no job at all, would stay unfinished forever. +func SettleRunAfterCancel(ctx context.Context, run *ActionRun) error { + if run.Status.IsDone() { + return nil + } + return refreshRunStatus(ctx, run.RepoID, run.ID, run.LatestAttemptID, StatusCancelled) +} + // cancelOneJob cancels a single job and returns the post-cancel row func cancelOneJob(ctx context.Context, job *ActionRunJob) (*ActionRunJob, error) { if job.Status.IsDone() { diff --git a/models/actions/run_job_test.go b/models/actions/run_job_test.go index 5bc91e4cf5c..182b19737f5 100644 --- a/models/actions/run_job_test.go +++ b/models/actions/run_job_test.go @@ -9,6 +9,7 @@ import ( "gitea.dev/models/db" "gitea.dev/models/unittest" + "gitea.dev/modules/timeutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -199,6 +200,94 @@ func TestCancelJobs_NestedBlockedReusableCaller(t *testing.T) { assert.Equal(t, StatusCancelled, gotRun.Status, "run must aggregate to Cancelled, not stay Blocked") } +func TestSettleRunAfterCancel(t *testing.T) { + // A run that cancelling updates no job in, because its jobs all reached a final status already + // or because it has none at all. Its own row has to be settled explicitly, or the run can never + // finish and can never be deleted either. + + newStuckRun := func(t *testing.T, withAttempt, withJob bool) (*ActionRun, []*ActionRunJob) { + t.Helper() + ctx := t.Context() + + run := &ActionRun{ + Title: "stuck-waiting", + RepoID: 4, + Index: 9801, + OwnerID: 1, + WorkflowID: "test.yaml", + TriggerUserID: 1, + Ref: "refs/heads/master", + CommitSHA: "c2d72f548424103f01ee1dc02889c1e2bff816b0", + Event: "push", + TriggerEvent: "push", + EventPayload: "{}", + Status: StatusWaiting, + } + require.NoError(t, db.Insert(ctx, run)) + + var runAttemptID int64 + if withAttempt { + attempt := &ActionRunAttempt{RepoID: run.RepoID, RunID: run.ID, Attempt: 1, TriggerUserID: 1, Status: StatusWaiting} + require.NoError(t, db.Insert(ctx, attempt)) + run.LatestAttemptID = attempt.ID + require.NoError(t, UpdateRun(ctx, run, "latest_attempt_id")) + runAttemptID = attempt.ID + } + + if !withJob { + return run, nil + } + job := &ActionRunJob{ + RunID: run.ID, + RunAttemptID: runAttemptID, + RepoID: run.RepoID, + OwnerID: run.OwnerID, + CommitSHA: run.CommitSHA, + Name: "job1", + JobID: "job1", + Attempt: 1, + Status: StatusSuccess, + Stopped: timeutil.TimeStampNow(), + } + require.NoError(t, db.Insert(ctx, job)) + return run, []*ActionRunJob{job} + } + + cases := []struct { + name string + withAttempt bool + withJob bool + want Status + }{ + {"done job", true, true, StatusSuccess}, + // Runs created before migration v331 have no attempt, their status lives on the run row itself. + {"done job on a legacy run without attempt", false, true, StatusSuccess}, + // Aggregation cannot reach a final status without any job, so cancelling has to end the run itself. + {"no job at all", true, false, StatusCancelled}, + {"no job at all on a legacy run without attempt", false, false, StatusCancelled}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + run, jobs := newStuckRun(t, tc.withAttempt, tc.withJob) + + // mirrors what the CancelRun service does + cancelled, err := CancelJobs(t.Context(), jobs) + require.NoError(t, err) + assert.Empty(t, cancelled, "nothing is cancellable, so the run row has to be settled explicitly") + require.NoError(t, SettleRunAfterCancel(t.Context(), run)) + + if tc.withAttempt { + gotAttempt := unittest.AssertExistsAndLoadBean(t, &ActionRunAttempt{ID: run.LatestAttemptID}) + assert.Equal(t, tc.want, gotAttempt.Status) + } + gotRun := unittest.AssertExistsAndLoadBean(t, &ActionRun{ID: run.ID}) + assert.Equal(t, tc.want, gotRun.Status) + assert.NotZero(t, gotRun.Stopped) + }) + } +} + func TestParseJobDeferredMatrixPlaceholder(t *testing.T) { // A placeholder is persisted with the raw matrix and without its needs, so routing its payload // through jobparser.Parse re-expands that matrix. The job emitter reads `if:` (and so ParseJob) diff --git a/services/actions/cancel.go b/services/actions/cancel.go index 8e7ae2a827b..458621e5a2a 100644 --- a/services/actions/cancel.go +++ b/services/actions/cancel.go @@ -19,7 +19,10 @@ func CancelRun(ctx context.Context, run *actions_model.ActionRun, jobs []*action if err != nil { return fmt.Errorf("CancelJobs: %w", err) } - return nil + if len(updatedJobs) > 0 { + return nil // a job update already refreshed the run + } + return actions_model.SettleRunAfterCancel(ctx, run) }); err != nil { return nil, err } @@ -27,14 +30,13 @@ func CancelRun(ctx context.Context, run *actions_model.ActionRun, jobs []*action CreateCommitStatusForRunJobs(ctx, run, jobs...) EmitJobsIfReadyByJobs(updatedJobs) NotifyWorkflowJobsStatusUpdate(ctx, updatedJobs...) - if len(updatedJobs) == 0 { - return run, nil - } reloaded, err := actions_model.GetRunByRepoAndID(ctx, run.RepoID, run.ID) if err != nil { return nil, fmt.Errorf("GetRunByRepoAndID: %w", err) } - NotifyWorkflowRunStatusUpdate(ctx, reloaded) + if len(updatedJobs) > 0 || reloaded.Status != run.Status { + NotifyWorkflowRunStatusUpdate(ctx, reloaded) + } return reloaded, nil }