From 6c8eaaf9e7b3fa86e1fdcc1434aa10fde665ab21 Mon Sep 17 00:00:00 2001 From: bircni Date: Sat, 31 Oct 2026 16:56:26 +0100 Subject: [PATCH] fix(actions): don't pick up jobs of repos with Actions disabled (#40106) (#40122) Backport https://github.com/go-gitea/gitea/pull/40106 Repository-level and instance-level runners still picked up waiting jobs of repositories with Actions disabled. https://github.com/go-gitea/gitea/pull/30331 only covered owner-level runners, by filtering them out in SQL. Pickup now checks the Actions unit of each job it is about to claim, for every runner scope, and cancels the job instead when Actions is disabled, also honoring `DISABLED_REPO_UNITS`. Disabling Actions therefore cancels its queued jobs once a matching runner polls, instead of leaving them waiting until `ABANDONED_JOB_TIMEOUT`. Co-authored-by: silverwind --- models/actions/run_job.go | 10 +--- models/actions/run_job_queue.go | 3 +- models/actions/task.go | 60 +++++++++---------- models/actions/task_test.go | 23 +++++-- services/actions/task.go | 6 +- services/actions/task_test.go | 4 +- .../actions_concurrent_claim_test.go | 8 +-- 7 files changed, 58 insertions(+), 56 deletions(-) diff --git a/models/actions/run_job.go b/models/actions/run_job.go index 7834c3dafba..39de8efaeb1 100644 --- a/models/actions/run_job.go +++ b/models/actions/run_job.go @@ -439,14 +439,6 @@ func CollectAllDescendantJobs(parent *ActionRunJob, allJobs []*ActionRunJob) []* return out } -// hasWaitingJobsToPick reports whether any waiting, unclaimed, non-reusable job -// remains in the repo, i.e. work that an idle runner could still pick up. -func hasWaitingJobsToPick(ctx context.Context, repoID int64) (bool, error) { - return db.GetEngine(ctx). - Where("repo_id = ? AND task_id = ? AND status = ? AND is_reusable_caller = ?", repoID, 0, StatusWaiting, false). - Exist(&ActionRunJob{}) -} - func UpdateRunJob(ctx context.Context, job *ActionRunJob, cond builder.Cond, cols ...string) (int64, error) { e := db.GetEngine(ctx) @@ -497,7 +489,7 @@ func UpdateRunJob(ctx context.Context, job *ActionRunJob, cond builder.Cond, col // Only bump when waiting work actually remains for this repo, otherwise // every job completion would needlessly bump the global version and wake // every idle runner instance-wide for nothing. - hasWaiting, err := hasWaitingJobsToPick(ctx, job.RepoID) + hasWaiting, err := JobQueueOptions{RepoID: job.RepoID, Status: StatusWaiting}.session(ctx).Exist(new(ActionRunJob)) if err != nil { return 0, err } diff --git a/models/actions/run_job_queue.go b/models/actions/run_job_queue.go index 86a5276b799..3dd2eb63243 100644 --- a/models/actions/run_job_queue.go +++ b/models/actions/run_job_queue.go @@ -27,13 +27,12 @@ func (opts JobQueueOptions) session(ctx context.Context) *xorm.Session { sess = sess.And(builder.Eq{"`action_run_job`.repo_id": opts.RepoID}) } if opts.OwnerID > 0 { - sess = sess.Join("INNER", "repository", "repository.id = `action_run_job`.repo_id AND repository.owner_id = ?", opts.OwnerID) + sess = sess.And(builder.In("`action_run_job`.repo_id", builder.Select("id").From("repository").Where(builder.Eq{"owner_id": opts.OwnerID}))) } return sess.And(opts.statusCond()) } var ( - // keep in sync with CreateTaskForRunner queuedJobsCond = builder.Eq{"`action_run_job`.status": StatusWaiting, "`action_run_job`.task_id": 0} // a cancelling job still occupies its runner runningJobsCond = builder.In("`action_run_job`.status", StatusRunning, StatusCancelling) diff --git a/models/actions/task.go b/models/actions/task.go index 418379c44c1..d0a76574fdc 100644 --- a/models/actions/task.go +++ b/models/actions/task.go @@ -261,21 +261,10 @@ var pickTaskBatchSize = 100 // atomically claims it. It iterates through all matching jobs so that a // concurrent claim by another runner (which would lose the optimistic lock on // job #1) does not leave the remaining jobs permanently unassigned. -func CreateTaskForRunner(ctx context.Context, runner *ActionRunner) (*ActionTask, bool, error) { +func CreateTaskForRunner(ctx context.Context, runner *ActionRunner) (task *ActionTask, cancelledJobs []*ActionRunJob, err error) { if db.InTransaction(ctx) { - return nil, false, errors.New("CreateTaskForRunner must not be called within a database transaction") + return nil, nil, errors.New("CreateTaskForRunner must not be called within a database transaction") } - e := db.GetEngine(ctx) - - jobCond := builder.NewCond() - if runner.RepoID != 0 { - jobCond = builder.Eq{"repo_id": runner.RepoID} - } else if runner.OwnerID != 0 { - jobCond = builder.In("repo_id", builder.Select("`repository`.id").From("repository"). - Join("INNER", "repo_unit", "`repository`.id = `repo_unit`.repo_id"). - Where(builder.Eq{"`repository`.owner_id": runner.OwnerID, "`repo_unit`.type": unit.TypeActions})) - } - 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) @@ -286,17 +275,17 @@ func CreateTaskForRunner(ctx context.Context, runner *ActionRunner) (*ActionTask var cursorUpdated timeutil.TimeStamp var cursorID int64 for { - cond := baseCond + sess := JobQueueOptions{RepoID: runner.RepoID, OwnerID: runner.OwnerID, Status: StatusWaiting}.session(ctx) if cursorID > 0 { - cond = cond.And(builder.Or( + sess.And(builder.Or( builder.Gt{"updated": cursorUpdated}, builder.And(builder.Eq{"updated": cursorUpdated}, builder.Gt{"id": cursorID}), )) } var jobs []*ActionRunJob - if err := e.Where(cond).Asc("updated", "id").Limit(pickTaskBatchSize).Find(&jobs); err != nil { - return nil, false, err + if err := sess.Asc("updated", "id").Limit(pickTaskBatchSize).Find(&jobs); err != nil { + return nil, cancelledJobs, err } // A short page means no waiting jobs remain beyond it. isLastPage := len(jobs) < pickTaskBatchSize @@ -309,28 +298,29 @@ func CreateTaskForRunner(ctx context.Context, runner *ActionRunner) (*ActionTask if !runner.CanMatchLabels(v.RunsOn) { continue } - task, ok, err := claimJobForRunner(ctx, runner, v) - if err != nil { - return nil, false, err + task, cancelled, err := claimJobForRunner(ctx, runner, v) + if cancelled { + cancelledJobs = append(cancelledJobs, v) } - if ok { - return task, true, nil + if err != nil || task != nil { + return task, cancelledJobs, err } - // Another runner claimed this job concurrently; try the next one. + // Another runner claimed this job concurrently, or it was cancelled. Try the next one. } if isLastPage { - return nil, false, nil + return nil, cancelledJobs, nil } } } // claimJobForRunner attempts to atomically claim job for runner inside its own -// transaction. Returns (task, true, nil) on success, or (nil, false, nil) when -// another runner wins the optimistic-lock race (the caller should try the next -// candidate job). +// transaction. Returns the task on success, cancelled when the job's repository +// has Actions disabled, or neither when another runner wins the optimistic-lock +// race (the caller should try the next candidate job). func claimJobForRunner(ctx context.Context, runner *ActionRunner, job *ActionRunJob) (*ActionTask, bool, error) { var resultTask *ActionTask + var cancelled bool err := db.WithTx(ctx, func(ctx context.Context) error { e := db.GetEngine(ctx) @@ -338,6 +328,15 @@ func claimJobForRunner(ctx context.Context, runner *ActionRunner, job *ActionRun if err := job.LoadAttributes(ctx); err != nil { return err } + if err := job.Run.Repo.LoadUnits(ctx); err != nil { + return err + } + if !job.Run.Repo.UnitEnabled(ctx, unit.TypeActions) { + job.Status, job.Stopped = StatusCancelled, timeutil.TimeStampNow() + affected, err := UpdateRunJob(ctx, job, builder.Eq{"task_id": 0, "status": StatusWaiting}, "status", "stopped") + cancelled = err == nil && affected == 1 + return err + } now := timeutil.TimeStampNow() job.Started = now @@ -403,13 +402,10 @@ func claimJobForRunner(ctx context.Context, runner *ActionRunner, job *ActionRun return nil }) - if errors.Is(err, errJobAlreadyClaimed) { - return nil, false, nil - } - if err != nil { + if err != nil && !errors.Is(err, errJobAlreadyClaimed) { return nil, false, err } - return resultTask, true, nil + return resultTask, cancelled, nil } // ReleaseTaskForRunner reverts a freshly-claimed but undelivered task: it deletes diff --git a/models/actions/task_test.go b/models/actions/task_test.go index d66ec1a3fc6..2119f3792b7 100644 --- a/models/actions/task_test.go +++ b/models/actions/task_test.go @@ -13,6 +13,7 @@ import ( runnerv1 "gitea.dev/actionslib/runner/v1" "gitea.dev/models/db" repo_model "gitea.dev/models/repo" + "gitea.dev/models/unit" "gitea.dev/models/unittest" "gitea.dev/modules/actions/jobparser" "gitea.dev/modules/timeutil" @@ -258,6 +259,11 @@ func TestReleaseTaskForRunner(t *testing.T) { } require.NoError(t, db.Insert(t.Context(), run)) + disabledRun := &ActionRun{RepoID: 2, TriggerUserID: 2} + require.NoError(t, db.Insert(t.Context(), disabledRun)) + disabledJob := &ActionRunJob{RunID: disabledRun.ID, RepoID: 2, Status: StatusWaiting, RunsOn: []string{"ubuntu-latest"}} + require.NoError(t, db.Insert(t.Context(), disabledJob)) + job := &ActionRunJob{ RunID: run.ID, RepoID: run.RepoID, @@ -279,11 +285,13 @@ func TestReleaseTaskForRunner(t *testing.T) { } runner.GenerateAndFillToken() require.NoError(t, db.Insert(t.Context(), runner)) + require.NoError(t, db.Insert(t.Context(), &repo_model.RepoUnit{RepoID: run.RepoID, Type: unit.TypeActions})) - task, ok, err := CreateTaskForRunner(t.Context(), runner) + task, cancelledJobs, err := CreateTaskForRunner(t.Context(), runner) require.NoError(t, err) - require.True(t, ok) require.NotNil(t, task) + require.Len(t, cancelledJobs, 1) + assert.Equal(t, StatusCancelled, unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: disabledJob.ID}).Status) claimed := unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: job.ID}) require.Equal(t, StatusRunning, claimed.Status) @@ -324,6 +332,11 @@ func TestCreateTaskForRunnerPagination(t *testing.T) { } require.NoError(t, db.Insert(t.Context(), run)) + disabledRun := &ActionRun{RepoID: 2, TriggerUserID: 2} + require.NoError(t, db.Insert(t.Context(), disabledRun)) + disabledJob := &ActionRunJob{RunID: disabledRun.ID, RepoID: 2, Status: StatusWaiting, RunsOn: []string{"ubuntu-latest"}} + require.NoError(t, db.Insert(t.Context(), disabledJob)) + // 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 { @@ -362,11 +375,13 @@ func TestCreateTaskForRunnerPagination(t *testing.T) { } runner.GenerateAndFillToken() require.NoError(t, db.Insert(t.Context(), runner)) + require.NoError(t, db.Insert(t.Context(), &repo_model.RepoUnit{RepoID: run.RepoID, Type: unit.TypeActions})) - task, ok, err := CreateTaskForRunner(t.Context(), runner) + task, cancelledJobs, err := CreateTaskForRunner(t.Context(), runner) require.NoError(t, err) - require.True(t, ok) require.NotNil(t, task) + require.Len(t, cancelledJobs, 1) + assert.Equal(t, StatusCancelled, unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: disabledJob.ID}).Status) claimed := unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: target.ID}) assert.Equal(t, StatusRunning, claimed.Status) diff --git a/services/actions/task.go b/services/actions/task.go index e5f39349124..b0794ade769 100644 --- a/services/actions/task.go +++ b/services/actions/task.go @@ -94,11 +94,13 @@ func PickTask(ctx context.Context, runner *actions_model.ActionRunner) (*runnerv } } - t, ok, err := actions_model.CreateTaskForRunner(ctx, runner) + t, cancelledJobs, err := actions_model.CreateTaskForRunner(ctx, runner) + NotifyWorkflowJobsAndRunsStatusUpdate(context.WithoutCancel(ctx), cancelledJobs) // the cancellations are committed even if the runner went away + EmitJobsIfReadyByJobs(cancelledJobs) if err != nil { return nil, false, fmt.Errorf("CreateTaskForRunner: %w", err) } - if !ok { + if t == nil { return nil, false, nil } diff --git a/services/actions/task_test.go b/services/actions/task_test.go index 397266059f3..89786ec608e 100644 --- a/services/actions/task_test.go +++ b/services/actions/task_test.go @@ -61,9 +61,9 @@ func TestReleaseTaskForRunnerCleanup(t *testing.T) { runner.GenerateAndFillToken() require.NoError(t, db.Insert(t.Context(), runner)) - task, ok, err := actions_model.CreateTaskForRunner(t.Context(), runner) + task, _, err := actions_model.CreateTaskForRunner(t.Context(), runner) require.NoError(t, err) - require.True(t, ok) + require.NotNil(t, task) require.Equal(t, actions_model.StatusRunning, unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: job.ID}).Status) // the cleanup helper uses its own context, so the claimed job is returned to the waiting queue diff --git a/tests/integration/actions_concurrent_claim_test.go b/tests/integration/actions_concurrent_claim_test.go index c097dbfd5bf..30f9cf0ba5d 100644 --- a/tests/integration/actions_concurrent_claim_test.go +++ b/tests/integration/actions_concurrent_claim_test.go @@ -89,15 +89,14 @@ func TestCreateTaskForRunnerConcurrentClaim(t *testing.T) { // as happens when all see the same stale tasksVersion simultaneously. type result struct { task *actions_model.ActionTask - ok bool err error } results := make([]result, numJobs) var wg sync.WaitGroup for i := range numJobs { wg.Go(func() { - task, ok, err := actions_model.CreateTaskForRunner(t.Context(), runners[i]) - results[i] = result{task, ok, err} + task, _, err := actions_model.CreateTaskForRunner(t.Context(), runners[i]) + results[i] = result{task, err} }) } wg.Wait() @@ -106,8 +105,7 @@ func TestCreateTaskForRunnerConcurrentClaim(t *testing.T) { claimedJobIDs := make(map[int64]bool) for i, r := range results { require.NoError(t, r.err, "runner %d got an unexpected error", i) - require.True(t, r.ok, "runner %d did not get a task even though free jobs exist", i) - require.NotNil(t, r.task) + require.NotNil(t, r.task, "runner %d did not get a task even though free jobs exist", i) assert.False(t, claimedJobIDs[r.task.JobID], "job %d was claimed by more than one runner", r.task.JobID) claimedJobIDs[r.task.JobID] = true }