Reduce unnecessary DB queries for Actions tasks (#219)

implement: https://github.com/go-gitea/gitea/issues/24544

Changes:
- Add a global variable `tasksVersion` to store the lastest version number which returned by Gitea.
- Pass `tasksVersion` to Gitea when invoking `FetchTask`.
- If there is no task in the `FetchTask` response, store the version from the `FetchTask` response into `tasksVersion` variable.
- If there is a task in the `FetchTask` response, set `tasksVersion` to zero to focre query db in next `FetchTask` request.

Related:
- Protocol: https://gitea.com/gitea/actions-proto-def/pulls/10
- Gitea side: https://github.com/go-gitea/gitea/pull/25199

Reviewed-on: https://gitea.com/gitea/act_runner/pulls/219
Co-authored-by: sillyguodong <gedong_1994@163.com>
Co-committed-by: sillyguodong <gedong_1994@163.com>

README.md

@@ -26,6 +26,13 @@ make docker
 
 ## Quickstart
 
+Actions are disabled by default, so you need to add the following to the configuration file of your Gitea instance to enable it: 
+  
+```ini
+[actions]
+ENABLED=true
+```
+
 ### Register
 
 ```bash
@@ -36,7 +43,7 @@ And you will be asked to input:
 
 1. Gitea instance URL, like `http://192.168.8.8:3000/`. You should use your gitea instance ROOT_URL as the instance argument
  and you should not use `localhost` or `127.0.0.1` as instance IP;
-2. Runner token, you can get it from `http://192.168.8.8:3000/admin/runners`;
+2. Runner token, you can get it from `http://192.168.8.8:3000/admin/actions/runners`;
 3. Runner name, you can just leave it blank;
 4. Runner labels, you can just leave it blank.
 

go.mod

@@ -3,7 +3,7 @@ module gitea.com/gitea/act_runner
 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.3-0.20230113022436-2b1561217fa5
 	github.com/avast/retry-go/v4 v4.3.1
 	github.com/bufbuild/connect-go v1.3.1

go.sum

@@ -1,5 +1,5 @@
-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.3-0.20230113022436-2b1561217fa5 h1:daBEK2GQeqGikJESctP5Cu1i33z5ztAD4kyQWiw185M=
 code.gitea.io/gitea-vet v0.2.3-0.20230113022436-2b1561217fa5/go.mod h1:zcNbT/aJEmivCAhfmkHOlT645KNOf9W2KnkLgFjGGfE=
 gitea.com/gitea/act v0.246.2-0.20230717034634-cdc6d4bc6a38 h1:whUEO/qPkYfpbL1he9TuIIzz2P4v6xEwb2lT6E/4F7A=

internal/app/poll/poller.go

@@ -6,7 +6,9 @@ package poll
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
+	"sync/atomic"
 
 	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
 	"github.com/bufbuild/connect-go"
@@ -19,9 +21,10 @@ import (
 )
 
 type Poller struct {
-	client client.Client
-	runner *run.Runner
-	cfg    *config.Config
+	client       client.Client
+	runner       *run.Runner
+	cfg          *config.Config
+	tasksVersion atomic.Int64 // tasksVersion used to store the version of the last task fetched from the Gitea.
 }
 
 func New(cfg *config.Config, client client.Client, runner *run.Runner) *Poller {
@@ -55,9 +58,20 @@ func (p *Poller) poll(ctx context.Context, wg *sync.WaitGroup, limiter *rate.Lim
 		if !ok {
 			continue
 		}
-		if err := p.runner.Run(ctx, task); err != nil {
-			log.WithError(err).Error("failed to run task")
+		p.runTaskWithRecover(ctx, task)
+	}
+}
+
+func (p *Poller) runTaskWithRecover(ctx context.Context, task *runnerv1.Task) {
+	defer func() {
+		if r := recover(); r != nil {
+			err := fmt.Errorf("panic: %v", r)
+			log.WithError(err).Error("panic in runTaskWithRecover")
 		}
+	}()
+
+	if err := p.runner.Run(ctx, task); err != nil {
+		log.WithError(err).Error("failed to run task")
 	}
 }
 
@@ -65,7 +79,11 @@ func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
 	reqCtx, cancel := context.WithTimeout(ctx, p.cfg.Runner.FetchTimeout)
 	defer cancel()
 
-	resp, err := p.client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{}))
+	// Load the version value that was in the cache when the request was sent.
+	v := p.tasksVersion.Load()
+	resp, err := p.client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{
+		TasksVersion: v,
+	}))
 	if errors.Is(err, context.DeadlineExceeded) {
 		err = nil
 	}
@@ -74,8 +92,20 @@ func (p *Poller) fetchTask(ctx context.Context) (*runnerv1.Task, bool) {
 		return nil, false
 	}
 
-	if resp == nil || resp.Msg == nil || resp.Msg.Task == nil {
+	if resp == nil || resp.Msg == nil {
 		return nil, false
 	}
+
+	if resp.Msg.TasksVersion > v {
+		p.tasksVersion.CompareAndSwap(v, resp.Msg.TasksVersion)
+	}
+
+	if resp.Msg.Task == nil {
+		return nil, false
+	}
+
+	// got a task, set `tasksVersion` to zero to focre query db in next request.
+	p.tasksVersion.CompareAndSwap(resp.Msg.TasksVersion, 0)
+
 	return resp.Msg.Task, true
 }