From f47e3d930d63750cf0f52e58fc75dfdc6c332cfa Mon Sep 17 00:00:00 2001 From: Giteabot Date: Sun, 26 Jul 2026 07:38:54 -0700 Subject: [PATCH] fix(actions): cancel tasks immediately when the runner stopped reporting (#38616) (#38644) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backport #38616 by @bircni 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: bircni Co-authored-by: Zettat123 Co-authored-by: silverwind --- models/actions/task.go | 15 +- models/actions/task_test.go | 299 +++++++++++++++--------------------- 2 files changed, 142 insertions(+), 172 deletions(-) diff --git a/models/actions/task.go b/models/actions/task.go index cfb0c5ec06..5a76409022 100644 --- a/models/actions/task.go +++ b/models/actions/task.go @@ -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 diff --git a/models/actions/task_test.go b/models/actions/task_test.go index aaec01156c..139de0c9ec 100644 --- a/models/actions/task_test.go +++ b/models/actions/task_test.go @@ -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 +}