perf(actions): debounce runner heartbeat writes and throttle task picks (#38281) (#38368)

Backport #38281 by @bircni

3 reductions in the DB load generated by many runners polling `FetchTask`:

**1. Debounce runner heartbeat writes**
Every poll wrote `last_online`, and every `UpdateTask`/`UpdateLog` wrote
`last_active` — while a runner streams logs that is many writes per
second per runner. These are now persisted only when stale enough to
actually affect the active/offline status (`ShouldPersistLastOnline` /
`ShouldPersistLastActive`), using the existing columns.

**2. Throttle concurrent task picks**
A new in-process semaphore (`MAX_CONCURRENT_TASK_PICKS`) bounds how many
runners run the task-assignment transaction at once, so a fleet polling
together cannot stampede the query. Throttled polls retry on their next
poll without advancing the runner's tasks version.

**3. Paginate the task-pick query**
`CreateTaskForRunner` previously loaded every waiting job in the
runner's scope into memory on each poll (no `LIMIT`). Now it pages
through the waiting backlog oldest-first with `LIMIT`, claiming the
first label-matching job.

Co-authored-by: bircni <bircni@icloud.com>
Co-authored-by: Zettat123 <zettat123@gmail.com>

custom/conf/app.example.ini

@@ -3008,6 +3008,10 @@ LEVEL = Info
 ;SCOPED_WORKFLOW_DIRS = .gitea/scoped_workflows
 ;; Maximum number of attempts a single workflow run can have. Default value is 50.
 ;MAX_RERUN_ATTEMPTS = 50
+;; Maximum number of runners that may run the task-assignment query concurrently, per Gitea instance.
+;; Caps this instance's DB load when many runners poll at once; excess runners retry on their next poll.
+;; In a multi-instance deployment the cluster-wide limit is this value times the number of instances. Default value is 16.
+;MAX_CONCURRENT_TASK_PICKS = 16
 
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

models/actions/runner.go

@@ -75,8 +75,28 @@ type ActionRunner struct {
 const (
 	RunnerOfflineTime = time.Minute
 	RunnerIdleTime    = 10 * time.Second
+	// RunnerHeartbeatInterval is how often last_online is persisted on poll.
+	// Must stay well below RunnerOfflineTime so runners don't flap to offline.
+	RunnerHeartbeatInterval = 30 * time.Second
+	// RunnerActiveInterval is how often last_active is persisted while a runner
+	// streams task updates and logs. Must stay well below RunnerIdleTime so a
+	// busy runner keeps showing ACTIVE without a DB write on every RPC.
+	RunnerActiveInterval = 5 * time.Second
 )
 
+// ShouldPersistLastOnline reports whether last_online is stale enough to be
+// worth writing back. Avoids a DB write on every runner poll.
+func ShouldPersistLastOnline(last timeutil.TimeStamp, now time.Time) bool {
+	return now.Sub(last.AsTime()) >= RunnerHeartbeatInterval
+}
+
+// ShouldPersistLastActive reports whether last_active is stale enough to be
+// worth writing back. Avoids a DB write on every UpdateTask/UpdateLog RPC while
+// a runner is actively streaming logs.
+func ShouldPersistLastActive(last timeutil.TimeStamp, now time.Time) bool {
+	return now.Sub(last.AsTime()) >= RunnerActiveInterval
+}
+
 // BelongsToOwnerName before calling, should guarantee that all attributes are loaded
 func (r *ActionRunner) BelongsToOwnerName() string {
 	if r.RepoID != 0 {

models/actions/runner_test.go

@@ -6,14 +6,86 @@ package actions
 import (
 	"fmt"
 	"testing"
+	"time"
 
 	"gitea.dev/models/db"
 	"gitea.dev/models/unittest"
+	"gitea.dev/modules/timeutil"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func TestShouldPersistLastOnline(t *testing.T) {
+	now := time.Now()
+	tests := []struct {
+		name string
+		last timeutil.TimeStamp
+		want bool
+	}{
+		{
+			name: "fresh, skip write",
+			last: timeutil.TimeStamp(now.Add(-5 * time.Second).Unix()),
+			want: false,
+		},
+		{
+			name: "exactly at interval, write",
+			last: timeutil.TimeStamp(now.Add(-RunnerHeartbeatInterval).Unix()),
+			want: true,
+		},
+		{
+			name: "stale, write",
+			last: timeutil.TimeStamp(now.Add(-2 * RunnerHeartbeatInterval).Unix()),
+			want: true,
+		},
+		{
+			name: "zero (never seen), write",
+			last: 0,
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, ShouldPersistLastOnline(tt.last, now))
+		})
+	}
+}
+
+func TestShouldPersistLastActive(t *testing.T) {
+	now := time.Now()
+	tests := []struct {
+		name string
+		last timeutil.TimeStamp
+		want bool
+	}{
+		{
+			name: "fresh, skip write",
+			last: timeutil.TimeStamp(now.Add(-1 * time.Second).Unix()),
+			want: false,
+		},
+		{
+			name: "exactly at interval, write",
+			last: timeutil.TimeStamp(now.Add(-RunnerActiveInterval).Unix()),
+			want: true,
+		},
+		{
+			name: "stale, write",
+			last: timeutil.TimeStamp(now.Add(-2 * RunnerActiveInterval).Unix()),
+			want: true,
+		},
+		{
+			name: "zero (never seen), write",
+			last: 0,
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, ShouldPersistLastActive(tt.last, now))
+		})
+	}
+}
+
 func TestFindRunnerOptions_ToOrders_StableTiebreaker(t *testing.T) {
 	// Sorts on a non-unique column must end with the unique id tiebreaker so
 	// pagination is deterministic; without it, runners sharing the same

models/actions/task.go

@@ -231,6 +231,11 @@ func makeTaskStepDisplayName(step *jobparser.Step, limit int) (name string) {
 // another runner won the optimistic-lock race; it is never returned to callers.
 var errJobAlreadyClaimed = errors.New("job already claimed by another runner")
 
+// pickTaskBatchSize bounds how many waiting jobs each CreateTaskForRunner query loads,
+// so a large backlog is not fetched into memory on every runner poll.
+// It is a var only so tests can shrink it to exercise pagination cheaply.
+var pickTaskBatchSize = 100
+
 // CreateTaskForRunner finds a waiting job that matches the runner's labels and
 // atomically claims it. It iterates through all matching jobs so that a
 // concurrent claim by another runner (which would lose the optimistic lock on
@@ -249,31 +254,51 @@ func CreateTaskForRunner(ctx context.Context, runner *ActionRunner) (*ActionTask
 			Join("INNER", "repo_unit", "`repository`.id = `repo_unit`.repo_id").
 			Where(builder.Eq{"`repository`.owner_id": runner.OwnerID, "`repo_unit`.type": unit.TypeActions}))
 	}
-	if jobCond.IsValid() {
-		jobCond = builder.In("run_id", builder.Select("id").From("action_run").Where(jobCond))
-	}
-
-	var jobs []*ActionRunJob
-	if err := e.Where("task_id=? AND status=? AND is_reusable_caller=?", 0, StatusWaiting, false).And(jobCond).Asc("updated", "id").Find(&jobs); err != nil {
-		return nil, false, err
-	}
+	baseCond := builder.Eq{"task_id": 0, "status": StatusWaiting, "is_reusable_caller": false}.And(jobCond)
 
 	// TODO: a more efficient way to filter labels
 	log.Trace("runner labels: %v", runner.AgentLabels)
-	for _, v := range jobs {
-		if !runner.CanMatchLabels(v.RunsOn) {
-			continue
+
+	// Page through the waiting jobs oldest-first instead of loading the whole backlog into memory on every poll.
+	// Keyset pagination on (updated, id) is safe under concurrent claims:
+	// updated only moves forward, so the advancing cursor never skips a still-waiting job even as claimed jobs drop out.
+	var cursorUpdated timeutil.TimeStamp
+	var cursorID int64
+	for {
+		cond := baseCond
+		if cursorID > 0 {
+			cond = cond.And(builder.Or(
+				builder.Gt{"updated": cursorUpdated},
+				builder.And(builder.Eq{"updated": cursorUpdated}, builder.Gt{"id": cursorID}),
+			))
 		}
-		task, ok, err := claimJobForRunner(ctx, runner, v)
-		if err != nil {
+
+		var jobs []*ActionRunJob
+		if err := e.Where(cond).Asc("updated", "id").Limit(pickTaskBatchSize).Find(&jobs); err != nil {
 			return nil, false, err
 		}
-		if ok {
-			return task, true, nil
+
+		for _, v := range jobs {
+			if !runner.CanMatchLabels(v.RunsOn) {
+				continue
+			}
+			task, ok, err := claimJobForRunner(ctx, runner, v)
+			if err != nil {
+				return nil, false, err
+			}
+			if ok {
+				return task, true, nil
+			}
+			// Another runner claimed this job concurrently; try the next one.
 		}
-		// Another runner claimed this job concurrently; try the next one.
+
+		// A short page means no waiting jobs remain beyond it.
+		if len(jobs) < pickTaskBatchSize {
+			return nil, false, nil
+		}
+		last := jobs[len(jobs)-1]
+		cursorUpdated, cursorID = last.Updated, last.ID
 	}
-	return nil, false, nil
 }
 
 // claimJobForRunner attempts to atomically claim job for runner inside its own

models/actions/task_test.go

@@ -371,3 +371,74 @@ func TestReleaseTaskForRunner(t *testing.T) {
 	unittest.AssertNotExistsBean(t, &ActionTask{ID: task.ID})
 	unittest.AssertNotExistsBean(t, &ActionTaskStep{TaskID: task.ID})
 }
+
+// TestCreateTaskForRunnerPagination verifies that a job sitting beyond the first page is still claimed
+func TestCreateTaskForRunnerPagination(t *testing.T) {
+	require.NoError(t, unittest.PrepareTestDatabase())
+
+	defer func(orig int) { pickTaskBatchSize = orig }(pickTaskBatchSize)
+	pickTaskBatchSize = 2
+
+	run := &ActionRun{
+		Title:         "pagination-test-run",
+		RepoID:        1,
+		OwnerID:       2,
+		WorkflowID:    "test.yaml",
+		Index:         9903,
+		TriggerUserID: 2,
+		Ref:           "refs/heads/main",
+		CommitSHA:     "c2d72f548424103f01ee1dc02889c1e2bff816b0",
+		Event:         "push",
+		TriggerEvent:  "push",
+		Status:        StatusWaiting,
+	}
+	require.NoError(t, db.Insert(t.Context(), run))
+
+	// Five waiting jobs the runner cannot run, then one it can.
+	// With a page size of 2 the matching job only appears on the third page.
+	for i := range 5 {
+		mismatch := &ActionRunJob{
+			RunID:     run.ID,
+			RepoID:    run.RepoID,
+			OwnerID:   run.OwnerID,
+			CommitSHA: run.CommitSHA,
+			Name:      "mismatch-" + string(rune('a'+i)),
+			Attempt:   1,
+			JobID:     "mismatch-" + string(rune('a'+i)),
+			Status:    StatusWaiting,
+			RunsOn:    []string{"windows-latest"},
+		}
+		require.NoError(t, db.Insert(t.Context(), mismatch))
+	}
+
+	target := &ActionRunJob{
+		RunID:           run.ID,
+		RepoID:          run.RepoID,
+		OwnerID:         run.OwnerID,
+		CommitSHA:       run.CommitSHA,
+		Name:            "target-job",
+		Attempt:         1,
+		JobID:           "target-job",
+		Status:          StatusWaiting,
+		RunsOn:          []string{"ubuntu-latest"},
+		WorkflowPayload: []byte("on: push\njobs:\n  target-job:\n    runs-on: ubuntu-latest\n    steps:\n      - run: echo hi\n"),
+	}
+	require.NoError(t, db.Insert(t.Context(), target))
+
+	runner := &ActionRunner{
+		UUID:        "pagination-runner-uuid",
+		Name:        "pagination-runner",
+		AgentLabels: []string{"ubuntu-latest"},
+	}
+	runner.GenerateAndFillToken()
+	require.NoError(t, db.Insert(t.Context(), runner))
+
+	task, ok, err := CreateTaskForRunner(t.Context(), runner)
+	require.NoError(t, err)
+	require.True(t, ok)
+	require.NotNil(t, task)
+
+	claimed := unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: target.ID})
+	assert.Equal(t, StatusRunning, claimed.Status)
+	assert.Equal(t, task.ID, claimed.TaskID)
+}

modules/setting/actions.go

@@ -14,6 +14,8 @@ import (
 
 const defaultMaxRerunAttempts = 50
 
+const defaultMaxConcurrentTaskPicks = 16
+
 // Actions settings
 var (
 	Actions = struct {
@@ -31,13 +33,18 @@ var (
 		WorkflowDirs          []string          `ini:"WORKFLOW_DIRS"`
 		ScopedWorkflowDirs    []string          `ini:"SCOPED_WORKFLOW_DIRS"`
 		MaxRerunAttempts      int64             `ini:"MAX_RERUN_ATTEMPTS"`
+		// MaxConcurrentTaskPicks bounds how many runners may run the task-assignment
+		// transaction at once per Gitea instance, to avoid a thundering herd when many
+		// runners poll together. It is a per-process limit, not a cluster-wide one.
+		MaxConcurrentTaskPicks int `ini:"MAX_CONCURRENT_TASK_PICKS"`
 	}{
-		Enabled:             true,
-		DefaultActionsURL:   defaultActionsURLGitHub,
-		SkipWorkflowStrings: []string{"[skip ci]", "[ci skip]", "[no ci]", "[skip actions]", "[actions skip]"},
-		WorkflowDirs:        []string{".gitea/workflows", ".github/workflows"},
-		ScopedWorkflowDirs:  []string{".gitea/scoped_workflows"},
-		MaxRerunAttempts:    defaultMaxRerunAttempts,
+		Enabled:                true,
+		DefaultActionsURL:      defaultActionsURLGitHub,
+		SkipWorkflowStrings:    []string{"[skip ci]", "[ci skip]", "[no ci]", "[skip actions]", "[actions skip]"},
+		WorkflowDirs:           []string{".gitea/workflows", ".github/workflows"},
+		ScopedWorkflowDirs:     []string{".gitea/scoped_workflows"},
+		MaxRerunAttempts:       defaultMaxRerunAttempts,
+		MaxConcurrentTaskPicks: defaultMaxConcurrentTaskPicks,
 	}
 )
 
@@ -128,6 +135,10 @@ func loadActionsFrom(rootCfg ConfigProvider) error {
 		Actions.MaxRerunAttempts = defaultMaxRerunAttempts
 	}
 
+	if Actions.MaxConcurrentTaskPicks <= 0 {
+		Actions.MaxConcurrentTaskPicks = defaultMaxConcurrentTaskPicks
+	}
+
 	if !Actions.LogCompression.IsValid() {
 		return fmt.Errorf("invalid [actions] LOG_COMPRESSION: %q", Actions.LogCompression)
 	}

routers/api/actions/runner/interceptor.go

@@ -8,6 +8,7 @@ import (
 	"crypto/subtle"
 	"errors"
 	"strings"
+	"time"
 
 	actions_model "gitea.dev/models/actions"
 	auth_model "gitea.dev/models/auth"
@@ -45,14 +46,26 @@ var withRunner = connect.WithInterceptors(connect.UnaryInterceptorFunc(func(unar
 			return nil, status.Error(codes.Unauthenticated, "unregistered runner")
 		}
 
-		cols := []string{"last_online"}
-		runner.LastOnline = timeutil.TimeStampNow()
-		if methodName == "UpdateTask" || methodName == "UpdateLog" {
-			runner.LastActive = timeutil.TimeStampNow()
+		now := time.Now()
+		cols := make([]string, 0, 2)
+		// Debounce last_active too: while a runner streams logs, UpdateLog fires
+		// many times per second and writing on each is a major source of DB load.
+		// Persist only when stale enough to affect the active/idle status.
+		if (methodName == "UpdateTask" || methodName == "UpdateLog") &&
+			actions_model.ShouldPersistLastActive(runner.LastActive, now) {
+			runner.LastActive = timeutil.TimeStamp(now.Unix())
 			cols = append(cols, "last_active")
 		}
-		if err := actions_model.UpdateRunner(ctx, runner, cols...); err != nil {
-			log.Error("can't update runner status: %v", err)
+		// Debounce last_online: writing on every poll is a major source of DB load
+		// with many runners. Persist only when stale enough to affect offline status.
+		if actions_model.ShouldPersistLastOnline(runner.LastOnline, now) {
+			runner.LastOnline = timeutil.TimeStamp(now.Unix())
+			cols = append(cols, "last_online")
+		}
+		if len(cols) > 0 {
+			if err := actions_model.UpdateRunner(ctx, runner, cols...); err != nil {
+				log.Error("can't update runner status: %v", err)
+			}
 		}
 
 		ctx = context.WithValue(ctx, runnerCtxKey{}, runner)

routers/api/actions/runner/runner.go

@@ -194,9 +194,15 @@ func (s *Service) FetchTask(
 		// if the task version in request is not equal to the version in db,
 		// it means there may still be some tasks that haven't been assigned.
 		// try to pick a task for the runner that send the request.
-		if t, ok, err := actions_service.PickTask(ctx, freshRunner); err != nil {
+		if t, ok, throttled, err := actions_service.TryPickTask(ctx, freshRunner); err != nil {
 			log.Error("pick task failed: %v", err)
 			return nil, status.Errorf(codes.Internal, "pick task: %v", err)
+		} else if throttled {
+			// Concurrency limit reached: don't advance the runner's tasks version,
+			// so it retries on its next poll instead of sleeping until the next bump.
+			latestVersion = tasksVersion
+			//  A steady stream here means MAX_CONCURRENT_TASK_PICKS is too low for the fleet.
+			log.Debug("task pick throttled for runner %q (id %d); it will retry on its next poll", freshRunner.Name, freshRunner.ID)
 		} else if ok {
 			task = t
 		}

services/actions/task.go

@@ -7,16 +7,47 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 
 	runnerv1 "gitea.dev/actions-proto-go/runner/v1"
 	actions_model "gitea.dev/models/actions"
 	"gitea.dev/models/db"
 	secret_model "gitea.dev/models/secret"
 	"gitea.dev/modules/log"
+	"gitea.dev/modules/setting"
 
 	"google.golang.org/protobuf/types/known/structpb"
 )
 
+var (
+	taskPickSem     chan struct{}
+	taskPickSemOnce sync.Once
+)
+
+func taskPickLimiter() chan struct{} {
+	taskPickSemOnce.Do(func() {
+		taskPickSem = make(chan struct{}, setting.Actions.MaxConcurrentTaskPicks)
+	})
+	return taskPickSem
+}
+
+// TryPickTask attempts to assign a task to the runner, bounding the number of
+// concurrent assignment transactions to avoid a thundering herd when many
+// runners poll at once. When the concurrency limit is reached it returns
+// throttled=true without touching the DB, so the caller can let the runner
+// retry on its next poll instead of advancing its tasks version.
+func TryPickTask(ctx context.Context, runner *actions_model.ActionRunner) (task *runnerv1.Task, ok, throttled bool, err error) {
+	sem := taskPickLimiter()
+	select {
+	case sem <- struct{}{}:
+		defer func() { <-sem }()
+	default:
+		return nil, false, true, nil
+	}
+	task, ok, err = PickTask(ctx, runner)
+	return task, ok, false, err
+}
+
 func PickTask(ctx context.Context, runner *actions_model.ActionRunner) (*runnerv1.Task, bool, error) {
 	var (
 		task       *runnerv1.Task

services/actions/task_test.go

@@ -0,0 +1,34 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package actions
+
+import (
+	"testing"
+
+	actions_model "gitea.dev/models/actions"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTryPickTaskThrottled(t *testing.T) {
+	sem := taskPickLimiter()
+
+	// Saturate every assignment slot so the next attempt must be throttled.
+	for range cap(sem) {
+		sem <- struct{}{}
+	}
+	defer func() {
+		for range cap(sem) {
+			<-sem
+		}
+	}()
+
+	// No DB access happens on the throttled path, so this is safe without fixtures.
+	task, ok, throttled, err := TryPickTask(t.Context(), &actions_model.ActionRunner{})
+	require.NoError(t, err)
+	assert.Nil(t, task)
+	assert.False(t, ok)
+	assert.True(t, throttled)
+}