fix(actions): cancel tasks immediately when the runner stopped reporting (#38616)

Fixes jobs that get stuck in `cancelling` after Gitea is restarted while a job is running. Reproduction: 1. Run a job with `sleep 100` 2. Stop Gitea and wait 120s (> 100s) 3. Start Gitea — the job is still `running`; cancel it, and it stays in `cancelling` ## Cause A cancellation is only ever delivered to a runner as the *response* to its `UpdateTask` RPC. A runner that has already given up on the task — it crashed, or it failed to report the final state while Gitea was unreachable — never calls `UpdateTask` again, so it never learns about the cancellation. The task then sits in `cancelling` until `stop_zombie_tasks` reaps it, which needs `ZOMBIE_TASK_TIMEOUT` (10m) of silence and only runs every 5 minutes. Cancelling actually made this worse. xorm rewrites the `updated` column on every `UPDATE`, even when `Cols()` restricts the update to `status`, so persisting the `cancelling` status reset the zombie clock. Pressing cancel pushed the cleanup a full `ZOMBIE_TASK_TIMEOUT` into the future instead of bringing it forward. ## Change `StopTask` now skips the `cancelling` handshake when the task has had no state report from its runner for longer than `TaskReportTimeout` (1 minute) and cancels it directly. This joins the two existing fallbacks — runner deleted, and runner without cancelling support — so every cancel path (web UI, API, concurrency, rerun) is covered. Runners report the state of a running task every few seconds, so a minute of silence means the runner is gone. The value is a constant rather than a setting because it only decides whether the runner is still reachable, not whether a task should be killed — `ZOMBIE_TASK_TIMEOUT` still owns that. ### Tradeoff If a runner is alive but has been silent for over a minute and is cancelled in that window, it skips the graceful post-step cleanup added in #37275. No work is lost: the runner still learns the outcome on its next report, because `UpdateTask` returns the task status as the result and the runner stops there. That is the behaviour that existed before #37275. --------- Co-authored-by: Zettat123 <zettat123@gmail.com> Co-authored-by: silverwind <me@silverwind.io>
2026-07-28 04:51:46 +02:00 · 2026-07-26 13:50:06 +00:00
parent 47565ae93e
commit 1b18b25c8e
2 changed files with 142 additions and 172 deletions
@@ -60,6 +60,12 @@ type ActionTask struct {
 	Updated timeutil.TimeStamp `xorm:"updated index"`
 }

+// taskReportTimeout is how long a task may go without contact from its runner before the
+// runner is assumed gone. Runners report state and stream logs every few seconds, both of
+// which refresh ActionTask.Updated. Shorter than setting.Actions.ZombieTaskTimeout because
+// it only decides whether the runner is reachable, not whether the task should be killed.
+const taskReportTimeout = time.Minute
+
 var successfulTokenTaskCache *lru.Cache[string, any]

 func init() {
@@ -567,6 +573,10 @@ func StopTask(ctx context.Context, taskID int64, status Status) error {
 			status = StatusCancelled
 		} else if !runner.HasCancellingSupport {
 			status = StatusCancelled
+		} else if task.Updated.AddDuration(taskReportTimeout) < now {
+			// A runner that stopped reporting will never acknowledge the cancellation either,
+			// so skip the handshake instead of waiting for the zombie task cleanup.
+			status = StatusCancelled
 		}
 	}

@@ -581,7 +591,10 @@ func StopTask(ctx context.Context, taskID int64, status Status) error {
 			return err
 		}

-		return UpdateTask(ctx, task, "status")
+		// NoAutoTime keeps "updated" at the runner's last contact: re-cancelling an already
+		// cancelling task must not defer the timeout above or the zombie task cleanup.
+		_, err := e.ID(task.ID).Cols("status").NoAutoTime().Update(task)
+		return err
 	}

 	task.Status = status
@@ -99,68 +99,11 @@ func TestMakeTaskStepDisplayName(t *testing.T) {
 }

 func TestTaskCancellingFinalizesToCancelled(t *testing.T) {
-	newRunningTask := func(t *testing.T) (*ActionTask, *ActionRunJob) {
-		t.Helper()
-
-		run := &ActionRun{
-			Title:         "cancelling-test-run",
-			RepoID:        1,
-			OwnerID:       2,
-			WorkflowID:    "test.yaml",
-			Index:         999,
-			TriggerUserID: 2,
-			Ref:           "refs/heads/master",
-			CommitSHA:     "c2d72f548424103f01ee1dc02889c1e2bff816b0",
-			Event:         "push",
-			TriggerEvent:  "push",
-			Status:        StatusRunning,
-			Started:       timeutil.TimeStampNow(),
-		}
-		require.NoError(t, db.Insert(t.Context(), run))
-
-		job := &ActionRunJob{
-			RunID:     run.ID,
-			RepoID:    run.RepoID,
-			OwnerID:   run.OwnerID,
-			CommitSHA: run.CommitSHA,
-			Name:      "cancelling-finalization-job",
-			Attempt:   1,
-			JobID:     "cancelling-finalization-job",
-			Status:    StatusRunning,
-		}
-		require.NoError(t, db.Insert(t.Context(), job))
-
-		runner := &ActionRunner{
-			UUID:                 "runner-cancelling-supported",
-			Name:                 "runner-cancelling-supported",
-			HasCancellingSupport: true,
-		}
-		require.NoError(t, db.Insert(t.Context(), runner))
-
-		task := &ActionTask{
-			JobID:     job.ID,
-			Attempt:   1,
-			RunnerID:  runner.ID,
-			Status:    StatusRunning,
-			Started:   timeutil.TimeStampNow(),
-			RepoID:    run.RepoID,
-			OwnerID:   run.OwnerID,
-			CommitSHA: run.CommitSHA,
-		}
-		require.NoError(t, db.Insert(t.Context(), task))
-
-		job.TaskID = task.ID
-		_, err := UpdateRunJob(t.Context(), job, nil, "task_id")
-		require.NoError(t, err)
-
-		return task, job
-	}
-
 	testResult := func(t *testing.T, result runnerv1.Result) {
 		t.Helper()
 		require.NoError(t, unittest.PrepareTestDatabase())

-		task, job := newRunningTask(t)
+		task, job := newRunningTaskForCancelling(t, "cancelling-finalization-job", true)
 		require.NoError(t, StopTask(t.Context(), task.ID, StatusCancelling))

 		taskAfterStop := unittest.AssertExistsAndLoadBean(t, &ActionTask{ID: task.ID})
@@ -190,137 +133,92 @@ func TestTaskCancellingFinalizesToCancelled(t *testing.T) {
 	})
 }

-func TestStopTaskCancellingFallsBackForLegacyRunner(t *testing.T) {
-	require.NoError(t, unittest.PrepareTestDatabase())
+// TestStopTaskCancellingFallsBackToCancelled covers the cases where the cancelling handshake can
+// never complete, so StopTask must cancel right away instead of waiting for the zombie task cleanup.
+func TestStopTaskCancellingFallsBackToCancelled(t *testing.T) {
+	assertCancelled := func(t *testing.T, task *ActionTask, job *ActionRunJob) {
+		t.Helper()

-	run := &ActionRun{
-		Title:         "cancelling-test-run",
-		RepoID:        1,
-		OwnerID:       2,
-		WorkflowID:    "test.yaml",
-		Index:         999,
-		TriggerUserID: 2,
-		Ref:           "refs/heads/master",
-		CommitSHA:     "c2d72f548424103f01ee1dc02889c1e2bff816b0",
-		Event:         "push",
-		TriggerEvent:  "push",
-		Status:        StatusRunning,
-		Started:       timeutil.TimeStampNow(),
+		taskAfterStop := unittest.AssertExistsAndLoadBean(t, &ActionTask{ID: task.ID})
+		assert.Equal(t, StatusCancelled, taskAfterStop.Status)
+		assert.NotZero(t, taskAfterStop.Stopped)
+
+		jobAfterStop := unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: job.ID})
+		assert.Equal(t, StatusCancelled, jobAfterStop.Status)
+		assert.NotZero(t, jobAfterStop.Stopped)
 	}
-	require.NoError(t, db.Insert(t.Context(), run))

-	job := &ActionRunJob{
-		RunID:     run.ID,
-		RepoID:    run.RepoID,
-		OwnerID:   run.OwnerID,
-		CommitSHA: run.CommitSHA,
-		Name:      "legacy-cancelling-job",
-		Attempt:   1,
-		JobID:     "legacy-cancelling-job",
-		Status:    StatusRunning,
-	}
-	require.NoError(t, db.Insert(t.Context(), job))
+	// A runner too old to know the cancelling state can only be stopped by a final status.
+	t.Run("legacy runner", func(t *testing.T) {
+		require.NoError(t, unittest.PrepareTestDatabase())
+		task, job := newRunningTaskForCancelling(t, "legacy-cancelling-job", false)

-	runner := &ActionRunner{
-		UUID:                 "runner-legacy-no-cancelling",
-		Name:                 "runner-legacy-no-cancelling",
-		HasCancellingSupport: false,
-	}
-	require.NoError(t, db.Insert(t.Context(), runner))
+		require.NoError(t, StopTask(t.Context(), task.ID, StatusCancelling))
+		assertCancelled(t, task, job)
+	})

-	task := &ActionTask{
-		JobID:     job.ID,
-		Attempt:   1,
-		RunnerID:  runner.ID,
-		Status:    StatusRunning,
-		Started:   timeutil.TimeStampNow(),
-		RepoID:    run.RepoID,
-		OwnerID:   run.OwnerID,
-		CommitSHA: run.CommitSHA,
-	}
-	require.NoError(t, db.Insert(t.Context(), task))
+	// The runner is gone, e.g. an ephemeral runner was cleaned up, so nobody can acknowledge.
+	t.Run("missing runner", func(t *testing.T) {
+		require.NoError(t, unittest.PrepareTestDatabase())
+		task, job := newRunningTaskForCancelling(t, "missing-runner-cancelling-job", true)

-	job.TaskID = task.ID
-	_, err := UpdateRunJob(t.Context(), job, nil, "task_id")
-	require.NoError(t, err)
+		_, err := db.DeleteByID[ActionRunner](t.Context(), task.RunnerID)
+		require.NoError(t, err)

-	require.NoError(t, StopTask(t.Context(), task.ID, StatusCancelling))
+		require.NoError(t, StopTask(t.Context(), task.ID, StatusCancelling))
+		assertCancelled(t, task, job)
+	})

-	taskAfterStop := unittest.AssertExistsAndLoadBean(t, &ActionTask{ID: task.ID})
-	assert.Equal(t, StatusCancelled, taskAfterStop.Status)
-	assert.NotZero(t, taskAfterStop.Stopped)
+	// The runner went silent, e.g. it gave up while Gitea was restarting, so it never picks up the request.
+	t.Run("silent runner", func(t *testing.T) {
+		require.NoError(t, unittest.PrepareTestDatabase())
+		task, job := newRunningTaskForCancelling(t, "silent-runner-cancelling-job", true)

-	jobAfterStop := unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: job.ID})
-	assert.Equal(t, StatusCancelled, jobAfterStop.Status)
-	assert.NotZero(t, jobAfterStop.Stopped)
+		// NoAutoTime because the point of the test is an "updated" older than xorm would write
+		task.Updated = timeutil.TimeStampNow().AddDuration(-2 * taskReportTimeout)
+		_, err := db.GetEngine(t.Context()).ID(task.ID).Cols("updated").NoAutoTime().Update(task)
+		require.NoError(t, err)
+
+		require.NoError(t, StopTask(t.Context(), task.ID, StatusCancelling))
+		assertCancelled(t, task, job)
+
+		// A runner coming back still learns the outcome from the UpdateTask response,
+		// and its late result does not overwrite the cancellation.
+		late, err := UpdateTaskByState(t.Context(), task.RunnerID, &runnerv1.TaskState{
+			Id:        task.ID,
+			Result:    runnerv1.Result_RESULT_SUCCESS,
+			StoppedAt: timestamppb.Now(),
+		})
+		require.NoError(t, err)
+		assert.Equal(t, StatusCancelled, late.Status)
+		assert.Equal(t, runnerv1.Result_RESULT_CANCELLED, late.Status.AsResult())
+	})
 }

-func TestStopTaskCancellingFallsBackForMissingRunner(t *testing.T) {
+// TestStopTaskCancellingKeepsReportTime makes sure persisting the cancelling status does not refresh
+// "updated": re-cancelling would otherwise defer both the fallback to cancelled and the zombie task
+// cleanup, no matter how long the runner has been silent.
+func TestStopTaskCancellingKeepsReportTime(t *testing.T) {
 	require.NoError(t, unittest.PrepareTestDatabase())
+	task, _ := newRunningTaskForCancelling(t, "repeated-cancelling-job", true)

-	run := &ActionRun{
-		Title:         "cancelling-test-run",
-		RepoID:        1,
-		OwnerID:       2,
-		WorkflowID:    "test.yaml",
-		Index:         999,
-		TriggerUserID: 2,
-		Ref:           "refs/heads/master",
-		CommitSHA:     "c2d72f548424103f01ee1dc02889c1e2bff816b0",
-		Event:         "push",
-		TriggerEvent:  "push",
-		Status:        StatusRunning,
-		Started:       timeutil.TimeStampNow(),
-	}
-	require.NoError(t, db.Insert(t.Context(), run))
-
-	job := &ActionRunJob{
-		RunID:     run.ID,
-		RepoID:    run.RepoID,
-		OwnerID:   run.OwnerID,
-		CommitSHA: run.CommitSHA,
-		Name:      "missing-runner-cancelling-job",
-		Attempt:   1,
-		JobID:     "missing-runner-cancelling-job",
-		Status:    StatusRunning,
-	}
-	require.NoError(t, db.Insert(t.Context(), job))
-
-	runner := &ActionRunner{
-		UUID:                 "runner-cleaned-up-before-cancel",
-		Name:                 "runner-cleaned-up-before-cancel",
-		HasCancellingSupport: true,
-	}
-	require.NoError(t, db.Insert(t.Context(), runner))
-
-	task := &ActionTask{
-		JobID:     job.ID,
-		Attempt:   1,
-		RunnerID:  runner.ID,
-		Status:    StatusRunning,
-		Started:   timeutil.TimeStampNow(),
-		RepoID:    run.RepoID,
-		OwnerID:   run.OwnerID,
-		CommitSHA: run.CommitSHA,
-	}
-	require.NoError(t, db.Insert(t.Context(), task))
-
-	job.TaskID = task.ID
-	_, err := UpdateRunJob(t.Context(), job, nil, "task_id")
-	require.NoError(t, err)
-
-	_, err = db.DeleteByID[ActionRunner](t.Context(), runner.ID)
+	// NoAutoTime because the point of the test is an "updated" older than xorm would write
+	lastReport := timeutil.TimeStampNow().AddDuration(-taskReportTimeout / 2)
+	task.Updated = lastReport
+	_, err := db.GetEngine(t.Context()).ID(task.ID).Cols("updated").NoAutoTime().Update(task)
 	require.NoError(t, err)

+	// the runner reported recently enough, so the task waits for it to acknowledge
 	require.NoError(t, StopTask(t.Context(), task.ID, StatusCancelling))
-
 	taskAfterStop := unittest.AssertExistsAndLoadBean(t, &ActionTask{ID: task.ID})
-	assert.Equal(t, StatusCancelled, taskAfterStop.Status)
-	assert.NotZero(t, taskAfterStop.Stopped)
+	assert.Equal(t, StatusCancelling, taskAfterStop.Status)
+	assert.Equal(t, lastReport, taskAfterStop.Updated)

-	jobAfterStop := unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: job.ID})
-	assert.Equal(t, StatusCancelled, jobAfterStop.Status)
-	assert.NotZero(t, jobAfterStop.Stopped)
+	// cancelling it again does not reset the clock either
+	require.NoError(t, StopTask(t.Context(), task.ID, StatusCancelling))
+	taskAfterSecondStop := unittest.AssertExistsAndLoadBean(t, &ActionTask{ID: task.ID})
+	assert.Equal(t, StatusCancelling, taskAfterSecondStop.Status)
+	assert.Equal(t, lastReport, taskAfterSecondStop.Updated)
 }

 // TestReleaseTaskForRunner verifies that releasing a freshly-claimed task returns
@@ -458,3 +356,62 @@ func TestCreateTaskForRunnerPagination(t *testing.T) {
 	assert.Equal(t, StatusRunning, claimed.Status)
 	assert.Equal(t, task.ID, claimed.TaskID)
 }
+
+// newRunningTaskForCancelling inserts a running run/job/task assigned to a fresh runner,
+// which is the state every cancellation test starts from.
+func newRunningTaskForCancelling(t *testing.T, name string, hasCancellingSupport bool) (*ActionTask, *ActionRunJob) {
+	t.Helper()
+
+	run := &ActionRun{
+		Title:         "cancelling-test-run",
+		RepoID:        1,
+		OwnerID:       2,
+		WorkflowID:    "test.yaml",
+		Index:         999,
+		TriggerUserID: 2,
+		Ref:           "refs/heads/master",
+		CommitSHA:     "c2d72f548424103f01ee1dc02889c1e2bff816b0",
+		Event:         "push",
+		TriggerEvent:  "push",
+		Status:        StatusRunning,
+		Started:       timeutil.TimeStampNow(),
+	}
+	require.NoError(t, db.Insert(t.Context(), run))
+
+	job := &ActionRunJob{
+		RunID:     run.ID,
+		RepoID:    run.RepoID,
+		OwnerID:   run.OwnerID,
+		CommitSHA: run.CommitSHA,
+		Name:      name,
+		Attempt:   1,
+		JobID:     name,
+		Status:    StatusRunning,
+	}
+	require.NoError(t, db.Insert(t.Context(), job))
+
+	runner := &ActionRunner{
+		UUID:                 name,
+		Name:                 name,
+		HasCancellingSupport: hasCancellingSupport,
+	}
+	require.NoError(t, db.Insert(t.Context(), runner))
+
+	task := &ActionTask{
+		JobID:     job.ID,
+		Attempt:   1,
+		RunnerID:  runner.ID,
+		Status:    StatusRunning,
+		Started:   timeutil.TimeStampNow(),
+		RepoID:    run.RepoID,
+		OwnerID:   run.OwnerID,
+		CommitSHA: run.CommitSHA,
+	}
+	require.NoError(t, db.Insert(t.Context(), task))
+
+	job.TaskID = task.ID
+	_, err := UpdateRunJob(t.Context(), job, nil, "task_id")
+	require.NoError(t, err)
+
+	return task, job
+}