fix: actions task state concurrent update (#38405) (#38409)

Backport #38405

fix #38333

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

models/actions/task.go

@@ -16,6 +16,7 @@ import (
 	"gitea.dev/models/db"
 	"gitea.dev/models/unit"
 	"gitea.dev/modules/actions/jobparser"
+	"gitea.dev/modules/globallock"
 	"gitea.dev/modules/log"
 	"gitea.dev/modules/setting"
 	"gitea.dev/modules/timeutil"
@@ -438,6 +439,18 @@ func UpdateTask(ctx context.Context, task *ActionTask, cols ...string) error {
 	return err
 }
 
+func getRunIDByTaskID(ctx context.Context, taskID int64) (runID int64, _ error) {
+	if has, err := db.GetEngine(ctx).Cols("action_run_job.run_id").
+		Table("action_task").
+		Join("INNER", "action_run_job", "action_run_job.id = action_task.job_id").
+		Where(builder.Eq{"action_task.id": taskID}).Get(&runID); err != nil {
+		return runID, err
+	} else if !has {
+		return runID, util.ErrNotExist
+	}
+	return runID, nil
+}
+
 // UpdateTaskByState updates the task by the state.
 // It will always update the task if the state is not final, even there is no change.
 // So it will update ActionTask.Updated to avoid the task being judged as a zombie task.
@@ -447,21 +460,26 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task
 		stepStates[v.Id] = v
 	}
 
-	return db.WithTx2(ctx, func(ctx context.Context) (*ActionTask, error) {
+	// Only one request can update the task because the final state needs to be calculated with all job states.
-		e := db.GetEngine(ctx)
+	// Otherwise, concurrent requests with transaction will make the SQL read stale job state and result in wrong final state.
-
+	taskID := state.Id
-		task := &ActionTask{}
+	runID, err := getRunIDByTaskID(ctx, taskID)
-		if has, err := e.ID(state.Id).Get(task); err != nil {
+	if err != nil {
-			return nil, err
+		return nil, err
+	}
+	task := &ActionTask{}
+	err = globallock.LockAndDo(ctx, fmt.Sprintf("UpdateTaskByState-run-%d", runID), func(ctx context.Context) error {
+		if has, err := db.GetEngine(ctx).ID(taskID).Get(task); err != nil {
+			return err
 		} else if !has {
-			return nil, util.ErrNotExist
+			return util.ErrNotExist
 		} else if runnerID != task.RunnerID {
-			return nil, errors.New("invalid runner for task")
+			return errors.New("invalid runner for task")
 		}
 
 		if task.Status.IsDone() {
 			// the state is final, do nothing
-			return task, nil
+			return nil
 		}
 
 		// state.Result is not unspecified means the task is finished
@@ -474,7 +492,7 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task
 			}
 			task.Stopped = timeutil.TimeStamp(state.StoppedAt.AsTime().Unix())
 			if err := UpdateTask(ctx, task, "status", "stopped"); err != nil {
-				return nil, err
+				return err
 			}
 			if _, err := UpdateRunJob(ctx, &ActionRunJob{
 				ID:      task.JobID,
@@ -482,18 +500,18 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task
 				Status:  task.Status,
 				Stopped: task.Stopped,
 			}, nil, "status", "stopped"); err != nil {
-				return nil, err
+				return err
 			}
 		} else {
 			// Force update ActionTask.Updated to avoid the task being judged as a zombie task
 			task.Updated = timeutil.TimeStampNow()
 			if err := UpdateTask(ctx, task, "updated"); err != nil {
-				return nil, err
+				return err
 			}
 		}
 
 		if err := task.LoadAttributes(ctx); err != nil {
-			return nil, err
+			return err
 		}
 
 		for _, step := range task.Steps {
@@ -510,13 +528,13 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task
 			} else if step.Started != 0 {
 				step.Status = StatusRunning
 			}
-			if _, err := e.ID(step.ID).Update(step); err != nil {
+			if _, err := db.GetEngine(ctx).ID(step.ID).Update(step); err != nil {
-				return nil, err
+				return err
 			}
 		}
-
+		return nil
-		return task, nil
 	})
+	return task, err
 }
 
 func StopTask(ctx context.Context, taskID int64, status Status) error {