fix(actions): allow cancelling runs without running jobs (#35842)

My idea is to allow cancelling an stuck run with all jobs done, in case such a bug happens again to not require admin commands. Related #35832 Co-authored-by: Zettat123 <zettat123@gmail.com> Co-authored-by: silverwind <me@silverwind.io>
2026-08-07 12:02:20 +00:00 · 2026-08-07 01:52:01 +02:00
parent 6ff3a65708
commit c210ef6dbb
3 changed files with 171 additions and 52 deletions
@@ -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() {
@@ -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)
@@ -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
 }