From 1bbd127a1a91c3107932001e54a2ab31bdc590e3 Mon Sep 17 00:00:00 2001 From: wxiaoguang Date: Sat, 11 Jul 2026 21:03:42 +0800 Subject: [PATCH] fix: actions task state concurrent update (#38405) fix #38333 --- models/actions/task.go | 52 ++++++++++++++++++++++++++++-------------- 1 file changed, 35 insertions(+), 17 deletions(-) diff --git a/models/actions/task.go b/models/actions/task.go index 4f810f576ee..047db27761c 100644 --- a/models/actions/task.go +++ b/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) { - e := db.GetEngine(ctx) - - task := &ActionTask{} - if has, err := e.ID(state.Id).Get(task); err != nil { - return nil, err + // Only one request can update the task because the final state needs to be calculated with all job states. + // Otherwise, concurrent requests with transaction will make the SQL read stale job state and result in wrong final state. + taskID := state.Id + runID, err := getRunIDByTaskID(ctx, taskID) + if err != nil { + 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 { - return nil, err + if _, err := db.GetEngine(ctx).ID(step.ID).Update(step); err != nil { + return err } } - - return task, nil + return nil }) + return task, err } func StopTask(ctx context.Context, taskID int64, status Status) error {