Add a transaction to pickTask (#33543)

In the old `pickTask`, when getting secrets or variables failed, the
task could get stuck in the `running` status (task status is `running`
but the runner did not fetch the task). To fix this issue, these steps
should be in one transaction.

---------

Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>

routers/api/actions/runner/runner.go

@@ -156,7 +156,7 @@ 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 not be assgined.
 		// try to pick a task for the runner that send the request.
-		if t, ok, err := pickTask(ctx, runner); err != nil {
+		if t, ok, err := actions_service.PickTask(ctx, runner); err != nil {
 			log.Error("pick task failed: %v", err)
 			return nil, status.Errorf(codes.Internal, "pick task: %v", err)
 		} else if ok {

services/actions/task.go

@@ -0,0 +1,107 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package actions
+
+import (
+	"context"
+	"fmt"
+
+	actions_model "code.gitea.io/gitea/models/actions"
+	"code.gitea.io/gitea/models/db"
+	secret_model "code.gitea.io/gitea/models/secret"
+
+	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func PickTask(ctx context.Context, runner *actions_model.ActionRunner) (*runnerv1.Task, bool, error) {
+	var (
+		task *runnerv1.Task
+		job  *actions_model.ActionRunJob
+	)
+
+	if err := db.WithTx(ctx, func(ctx context.Context) error {
+		t, ok, err := actions_model.CreateTaskForRunner(ctx, runner)
+		if err != nil {
+			return fmt.Errorf("CreateTaskForRunner: %w", err)
+		}
+		if !ok {
+			return nil
+		}
+
+		if err := t.LoadAttributes(ctx); err != nil {
+			return fmt.Errorf("task LoadAttributes: %w", err)
+		}
+		job = t.Job
+
+		secrets, err := secret_model.GetSecretsOfTask(ctx, t)
+		if err != nil {
+			return fmt.Errorf("GetSecretsOfTask: %w", err)
+		}
+
+		vars, err := actions_model.GetVariablesOfRun(ctx, t.Job.Run)
+		if err != nil {
+			return fmt.Errorf("GetVariablesOfRun: %w", err)
+		}
+
+		needs, err := findTaskNeeds(ctx, job)
+		if err != nil {
+			return fmt.Errorf("findTaskNeeds: %w", err)
+		}
+
+		taskContext, err := generateTaskContext(t)
+		if err != nil {
+			return fmt.Errorf("generateTaskContext: %w", err)
+		}
+
+		task = &runnerv1.Task{
+			Id:              t.ID,
+			WorkflowPayload: t.Job.WorkflowPayload,
+			Context:         taskContext,
+			Secrets:         secrets,
+			Vars:            vars,
+			Needs:           needs,
+		}
+
+		return nil
+	}); err != nil {
+		return nil, false, err
+	}
+
+	if task == nil {
+		return nil, false, nil
+	}
+
+	CreateCommitStatus(ctx, job)
+
+	return task, true, nil
+}
+
+func generateTaskContext(t *actions_model.ActionTask) (*structpb.Struct, error) {
+	giteaRuntimeToken, err := CreateAuthorizationToken(t.ID, t.Job.RunID, t.JobID)
+	if err != nil {
+		return nil, err
+	}
+
+	gitCtx := GenerateGiteaContext(t.Job.Run, t.Job)
+	gitCtx["token"] = t.Token
+	gitCtx["gitea_runtime_token"] = giteaRuntimeToken
+
+	return structpb.NewStruct(gitCtx)
+}
+
+func findTaskNeeds(ctx context.Context, taskJob *actions_model.ActionRunJob) (map[string]*runnerv1.TaskNeed, error) {
+	taskNeeds, err := FindTaskNeeds(ctx, taskJob)
+	if err != nil {
+		return nil, err
+	}
+	ret := make(map[string]*runnerv1.TaskNeed, len(taskNeeds))
+	for jobID, taskNeed := range taskNeeds {
+		ret[jobID] = &runnerv1.TaskNeed{
+			Outputs: taskNeed.Outputs,
+			Result:  runnerv1.Result(taskNeed.Result),
+		}
+	}
+	return ret, nil
+}