From 64d559a8e65100ae071deda44de083e88f15f2cf Mon Sep 17 00:00:00 2001 From: Giteabot Date: Tue, 7 Jul 2026 16:27:00 -0700 Subject: [PATCH] perf(actions): debounce runner heartbeat writes and throttle task picks (#38281) (#38368) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-authored-by: Zettat123 --- custom/conf/app.example.ini | 4 ++ models/actions/runner.go | 20 +++++++ models/actions/runner_test.go | 72 +++++++++++++++++++++++ models/actions/task.go | 59 +++++++++++++------ models/actions/task_test.go | 71 ++++++++++++++++++++++ modules/setting/actions.go | 23 ++++++-- routers/api/actions/runner/interceptor.go | 25 ++++++-- routers/api/actions/runner/runner.go | 8 ++- services/actions/task.go | 31 ++++++++++ services/actions/task_test.go | 34 +++++++++++ 10 files changed, 317 insertions(+), 30 deletions(-) create mode 100644 services/actions/task_test.go diff --git a/custom/conf/app.example.ini b/custom/conf/app.example.ini index e5451af76a..2e3787cbbd 100644 --- a/custom/conf/app.example.ini +++ b/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 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; diff --git a/models/actions/runner.go b/models/actions/runner.go index 593ebd5301..bc4367b52e 100644 --- a/models/actions/runner.go +++ b/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 { diff --git a/models/actions/runner_test.go b/models/actions/runner_test.go index 15661802fb..47b8a69307 100644 --- a/models/actions/runner_test.go +++ b/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 diff --git a/models/actions/task.go b/models/actions/task.go index 27b7efe124..4f810f576e 100644 --- a/models/actions/task.go +++ b/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 diff --git a/models/actions/task_test.go b/models/actions/task_test.go index e028ad5106..05f564f2b4 100644 --- a/models/actions/task_test.go +++ b/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) +} diff --git a/modules/setting/actions.go b/modules/setting/actions.go index 69e847f80f..6043d53ae5 100644 --- a/modules/setting/actions.go +++ b/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) } diff --git a/routers/api/actions/runner/interceptor.go b/routers/api/actions/runner/interceptor.go index a3508aa9b1..ec1ba58e91 100644 --- a/routers/api/actions/runner/interceptor.go +++ b/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) diff --git a/routers/api/actions/runner/runner.go b/routers/api/actions/runner/runner.go index 012f875ec7..44e0151c86 100644 --- a/routers/api/actions/runner/runner.go +++ b/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 } diff --git a/services/actions/task.go b/services/actions/task.go index f9e8e2d018..b063bff331 100644 --- a/services/actions/task.go +++ b/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 diff --git a/services/actions/task_test.go b/services/actions/task_test.go new file mode 100644 index 0000000000..f47bc93655 --- /dev/null +++ b/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) +}