Reduce unnecessary DB queries for Actions tasks

go.mod

@@ -3,7 +3,7 @@ module code.gitea.io/gitea
 go 1.20
 
 require (
-	code.gitea.io/actions-proto-go v0.3.0
+	code.gitea.io/actions-proto-go v0.3.1
 	code.gitea.io/gitea-vet v0.2.2
 	code.gitea.io/sdk/gitea v0.15.1
 	codeberg.org/gusted/mcaptcha v0.0.0-20220723083913-4f3072e1d570

go.sum

@@ -40,8 +40,8 @@ cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohl
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
 cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3fOKtUw0Xmo=
-code.gitea.io/actions-proto-go v0.3.0 h1:9Tvg8+TaaCXPKi6EnWl9vVgs2VZsj1Cs5afnsHa4AmM=
-code.gitea.io/actions-proto-go v0.3.0/go.mod h1:00ys5QDo1iHN1tHNvvddAcy2W/g+425hQya1cCSvq9A=
+code.gitea.io/actions-proto-go v0.3.1 h1:PMyiQtBKb8dNnpEO2R5rcZdXSis+UQZVo/SciMtR1aU=
+code.gitea.io/actions-proto-go v0.3.1/go.mod h1:00ys5QDo1iHN1tHNvvddAcy2W/g+425hQya1cCSvq9A=
 code.gitea.io/gitea-vet v0.2.1/go.mod h1:zcNbT/aJEmivCAhfmkHOlT645KNOf9W2KnkLgFjGGfE=
 code.gitea.io/gitea-vet v0.2.2 h1:TEOV/Glf38iGmKzKP0EB++Z5OSL4zGg3RrAvlwaMuvk=
 code.gitea.io/gitea-vet v0.2.2/go.mod h1:zcNbT/aJEmivCAhfmkHOlT645KNOf9W2KnkLgFjGGfE=

models/actions/tasks_version.go

@@ -0,0 +1,81 @@
+// Copyright 2023 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package actions
+
+import (
+	"context"
+	"fmt"
+
+	"code.gitea.io/gitea/models/db"
+	"code.gitea.io/gitea/modules/log"
+	"code.gitea.io/gitea/modules/timeutil"
+	"code.gitea.io/gitea/modules/util"
+)
+
+type ActionTasksVersion struct {
+	ID          int64 `xorm:"pk autoincr"`
+	OwnerID     int64 `xorm:"UNIQUE(owner_repo)"`
+	RepoID      int64 `xorm:"INDEX UNIQUE(owner_repo)"`
+	Version     int64
+	CreatedUnix timeutil.TimeStamp `xorm:"created"`
+	UpdatedUnix timeutil.TimeStamp `xorm:"updated"`
+}
+
+func init() {
+	db.RegisterModel(new(ActionTasksVersion))
+}
+
+func GetTasksVersionByScope(ctx context.Context, ownerID, repoID int64) (*ActionTasksVersion, error) {
+	var tasksVersion ActionTasksVersion
+	has, err := db.GetEngine(ctx).Where("owner_id = ? AND repo_id = ?", ownerID, repoID).Get(&tasksVersion)
+	if err != nil {
+		return nil, err
+	} else if !has {
+		return nil, fmt.Errorf("tasks version with owner id %d repo id %d: %w", ownerID, repoID, util.ErrNotExist)
+	}
+	return &tasksVersion, err
+}
+
+func InsertTasksVersion(ctx context.Context, ownerID, repoID int64) (*ActionTasksVersion, error) {
+	tasksVersion := &ActionTasksVersion{
+		OwnerID: ownerID,
+		RepoID:  repoID,
+		// Set the default value of version to 1, so that the first fetch request after the runner starts will definitely query the database.
+		Version: 1,
+	}
+	return tasksVersion, db.Insert(ctx, tasksVersion)
+}
+
+func increaseTasksVersionByScope(ctx context.Context, ownerID, repoID int64) (bool, error) {
+	result, err := db.GetEngine(ctx).Exec("UPDATE action_tasks_version SET version = version + 1 WHERE owner_id = ? AND repo_id = ?", ownerID, repoID)
+	if err != nil {
+		return false, err
+	}
+	affected, err := result.RowsAffected()
+	if err != nil {
+		return false, err
+	}
+
+	return affected != 0, err
+}
+
+func IncreaseTaskVersion(ctx context.Context, ownerID, repoID int64) error {
+	// increase tasks version
+	// 1. increase global
+	if _, err := increaseTasksVersionByScope(ctx, 0, 0); err != nil {
+		log.Error("IncreaseTasksVersionByScope(Global): %v", err)
+		return err
+	}
+	// 2. increase owner
+	if _, err := increaseTasksVersionByScope(ctx, ownerID, 0); err != nil {
+		log.Error("IncreaseTasksVersionByScope(Owner): %v", err)
+		return err
+	}
+	// 3. increase repo
+	if _, err := increaseTasksVersionByScope(ctx, 0, repoID); err != nil {
+		log.Error("IncreaseTasksVersionByScope(Repo): %v", err)
+		return err
+	}
+	return nil
+}

models/migrations/migrations.go

@@ -511,6 +511,8 @@ var migrations = []Migration{
 	NewMigration("Add git_size and lfs_size columns to repository table", v1_21.AddGitSizeAndLFSSizeToRepositoryTable),
 	// v264 -> v265
 	NewMigration("Add branch table", v1_21.AddBranchTable),
+	// v265 -> v266
+	NewMigration("Add action_tasks_version table", v1_21.CreateActionTasksVersionTable),
 }
 
 // GetCurrentDBVersion returns the current db version

models/migrations/v1_21/v265.go

@@ -0,0 +1,61 @@
+// Copyright 2023 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package v1_21 //nolint
+
+import (
+	"code.gitea.io/gitea/modules/timeutil"
+
+	"xorm.io/xorm"
+)
+
+func CreateActionTasksVersionTable(x *xorm.Engine) error {
+	sess := x.NewSession()
+	defer sess.Close()
+
+	if err := sess.Begin(); err != nil {
+		return err
+	}
+
+	type ActionTasksVersion struct {
+		ID          int64 `xorm:"pk autoincr"`
+		OwnerID     int64 `xorm:"UNIQUE(owner_repo)"`
+		RepoID      int64 `xorm:"INDEX UNIQUE(owner_repo)"`
+		Version     int64
+		CreatedUnix timeutil.TimeStamp `xorm:"created"`
+		UpdatedUnix timeutil.TimeStamp `xorm:"updated"`
+	}
+
+	// cerate action_tasks_version table.
+	if err := sess.Sync(new(ActionTasksVersion)); err != nil {
+		return err
+	}
+
+	// initialize data
+	type ScopeItem struct {
+		OwnerID int64
+		RepoID  int64
+	}
+	scopes := []ScopeItem{}
+	if err := sess.Distinct("owner_id", "repo_id").Table("action_runner").Where("deleted is null").Find(&scopes); err != nil {
+		return err
+	}
+
+	if len(scopes) > 0 {
+		versions := make([]ActionTasksVersion, 0, len(scopes))
+		for _, scope := range scopes {
+			versions = append(versions, ActionTasksVersion{
+				OwnerID: scope.OwnerID,
+				RepoID:  scope.RepoID,
+				// Set the default value of version to 1, so that the first fetch request after the runner starts will definitely query the database.
+				Version: 1,
+			})
+		}
+
+		if _, err := sess.Insert(&versions); err != nil {
+			return err
+		}
+	}
+
+	return sess.Commit()
+}

routers/api/actions/runner/runner.go

@@ -87,6 +87,16 @@ func (s *Service) Register(
 		return nil, errors.New("can't update runner token status")
 	}
 
+	if _, err := actions_model.GetTasksVersionByScope(ctx, runner.OwnerID, runner.RepoID); err != nil {
+		if !errors.Is(err, util.ErrNotExist) {
+			return nil, errors.New("query tasks version failure")
+		}
+		// create a row of action_tasks_version if not exists yet.
+		if _, err := actions_model.InsertTasksVersion(ctx, runner.OwnerID, runner.RepoID); err != nil {
+			return nil, errors.New("can't insert tasks version")
+		}
+	}
+
 	res := connect.NewResponse(&runnerv1.RegisterResponse{
 		Runner: &runnerv1.Runner{
 			Id:      runner.ID,
@@ -127,20 +137,33 @@ func (s *Service) Declare(
 // FetchTask assigns a task to the runner
 func (s *Service) FetchTask(
 	ctx context.Context,
-	_ *connect.Request[runnerv1.FetchTaskRequest],
+	req *connect.Request[runnerv1.FetchTaskRequest],
 ) (*connect.Response[runnerv1.FetchTaskResponse], error) {
 	runner := GetRunner(ctx)
 
 	var task *runnerv1.Task
-	if t, ok, err := pickTask(ctx, runner); err != nil {
-		log.Error("pick task failed: %v", err)
-		return nil, status.Errorf(codes.Internal, "pick task: %v", err)
-	} else if ok {
-		task = t
+	tasksVersion := req.Msg.TasksVersion // task version from runner
+	atv, err := actions_model.GetTasksVersionByScope(ctx, runner.OwnerID, runner.RepoID)
+	if err != nil {
+		log.Error("query tasks version failed: %v", err)
+		return nil, status.Errorf(codes.Internal, "query tasks version: %v", err)
+	}
+	latestVersion := atv.Version
+	if req.Msg.TasksVersion != latestVersion {
+		// if the task version in request is not equal to the version in db,
+		// it means there may still be some tasks not be assgined.
+		// try to pick a task for the runner that send the request.
+		if t, ok, err := pickTask(ctx, runner); err != nil {
+			log.Error("pick task failed: %v", err)
+			return nil, status.Errorf(codes.Internal, "pick task: %v", err)
+		} else if ok {
+			task = t
+		}
+		tasksVersion = latestVersion
 	}
-
 	res := connect.NewResponse(&runnerv1.FetchTaskResponse{
-		Task: task,
+		Task:         task,
+		TasksVersion: tasksVersion,
 	})
 	return res, nil
 }

routers/web/repo/actions/view.go

@@ -268,6 +268,11 @@ func RerunOne(ctx *context_module.Context) {
 		return
 	}
 
+	if err := actions_model.IncreaseTaskVersion(ctx, job.OwnerID, job.RepoID); err != nil {
+		ctx.Error(http.StatusInternalServerError, err.Error())
+		return
+	}
+
 	ctx.JSON(http.StatusOK, struct{}{})
 }
 
@@ -286,6 +291,11 @@ func RerunAll(ctx *context_module.Context) {
 		}
 	}
 
+	if err := actions_model.IncreaseTaskVersion(ctx, jobs[0].OwnerID, jobs[0].RepoID); err != nil {
+		ctx.Error(http.StatusInternalServerError, err.Error())
+		return
+	}
+
 	ctx.JSON(http.StatusOK, struct{}{})
 }
 

services/actions/notifier_helper.go

@@ -241,8 +241,13 @@ func notify(ctx context.Context, input *notifyInput) error {
 		} else {
 			CreateCommitStatus(ctx, jobs...)
 		}
+	}
 
+	if err := actions_model.IncreaseTaskVersion(ctx, input.Repo.OwnerID, input.Repo.ID); err != nil {
+		log.Error("IncreaseTaskVersion: %v", err)
+		return err
 	}
+
 	return nil
 }