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 <me@silverwind.io>

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
 			}

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)

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

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)

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
 	}
 

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

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
 	}