feat: cancel abandoned jobs

models/bots/run_job.go

@@ -25,12 +25,12 @@ type RunJob struct {
 	Ready           bool // ready to be executed
 	Attempt         int64
 	WorkflowPayload []byte
-	JobID           string   // job id in workflow, not job's id
+	JobID           string             // job id in workflow, not job's id
-	Needs           []int64  `xorm:"JSON TEXT"`
+	Needs           []int64            `xorm:"JSON TEXT"`
-	RunsOn          []string `xorm:"JSON TEXT"`
+	RunsOn          []string           `xorm:"JSON TEXT"`
-	TaskID          int64    // the latest task of the job
+	TaskID          int64              // the latest task of the job
-	Status          Status   `xorm:"index"`
+	Status          Status             `xorm:"index"`
-	Started         timeutil.TimeStamp
+	Started         timeutil.TimeStamp `xorm:"index"`
 	Stopped         timeutil.TimeStamp
 	Created         timeutil.TimeStamp `xorm:"created"`
 	Updated         timeutil.TimeStamp `xorm:"updated"`

models/bots/run_job_list.go

@@ -0,0 +1,46 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package bots
+
+import (
+	"context"
+
+	"code.gitea.io/gitea/models/db"
+	"code.gitea.io/gitea/modules/timeutil"
+	"xorm.io/builder"
+)
+
+type RunJobList []*RunJob
+
+type FindRunJobOptions struct {
+	db.ListOptions
+	Status        Status
+	StartedBefore timeutil.TimeStamp
+}
+
+func (opts FindRunJobOptions) toConds() builder.Cond {
+	cond := builder.NewCond()
+	if opts.Status > StatusUnknown {
+		cond = cond.And(builder.Eq{"status": opts.Status})
+	}
+	if opts.StartedBefore > 0 {
+		cond = cond.And(builder.Lt{"started": opts.StartedBefore})
+	}
+	return cond
+}
+
+func FindRunJobs(ctx context.Context, opts FindRunJobOptions) (RunJobList, int64, error) {
+	e := db.GetEngine(ctx).Where(opts.toConds())
+	if opts.PageSize > 0 && opts.Page >= 1 {
+		e.Limit(opts.PageSize, (opts.Page-1)*opts.PageSize)
+	}
+	var tasks RunJobList
+	total, err := e.FindAndCount(&tasks)
+	return tasks, total, err
+}
+
+func CountRunJobs(ctx context.Context, opts FindRunJobOptions) (int64, error) {
+	return db.GetEngine(ctx).Where(opts.toConds()).Count(new(RunJob))
+}

services/bots/clear_tasks.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	bots_model "code.gitea.io/gitea/models/bots"
+	"code.gitea.io/gitea/models/db"
 	"code.gitea.io/gitea/modules/log"
 	"code.gitea.io/gitea/modules/timeutil"
 	runnerv1 "gitea.com/gitea/proto-go/runner/v1"
@@ -62,6 +63,27 @@ func StopEndlessTasks(ctx context.Context) error {
 
 // CancelAbandonedJobs cancels the jobs which have waiting status, but haven't been picked by a runner for a long time
 func CancelAbandonedJobs(ctx context.Context) error {
-	// TODO
+	jobs, _, err := bots_model.FindRunJobs(ctx, bots_model.FindRunJobOptions{
+		Status:        bots_model.StatusWaiting,
+		StartedBefore: timeutil.TimeStamp(time.Now().Add(-abandonedJobTimeout).Unix()),
+	})
+	if err != nil {
+		log.Warn("find abandoned tasks: %v", err)
+		return err
+	}
+
+	now := timeutil.TimeStampNow()
+	for _, job := range jobs {
+		job.Status = bots_model.StatusCancelled
+		job.Stopped = now
+		if err := db.WithTx(func(ctx context.Context) error {
+			_, err := bots_model.UpdateRunJob(ctx, job, nil, "status", "stopped")
+			return err
+		}, ctx); err != nil {
+			log.Warn("cancel abandoned job %v: %v", job.ID, err)
+			// go on
+		}
+	}
+
 	return nil
 }