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 }