mirror of
https://github.com/go-gitea/gitea.git
synced 2026-07-28 01:05:20 +02:00
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 <bircni@icloud.com> Co-authored-by: Zettat123 <zettat123@gmail.com> Co-authored-by: silverwind <me@silverwind.io>
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user