Use db.WithTx/WithTx2 instead of TxContext when possible

models/actions/run.go

@@ -282,77 +282,72 @@ func CancelPreviousJobs(ctx context.Context, repoID int64, ref, workflowID strin
 // InsertRun inserts a run
 // The title will be cut off at 255 characters if it's longer than 255 characters.
 func InsertRun(ctx context.Context, run *ActionRun, jobs []*jobparser.SingleWorkflow) error {
-ctx, committer, err := db.TxContext(ctx)
-if err != nil {
-return err
-}
-defer committer.Close()
-
-index, err := db.GetNextResourceIndex(ctx, "action_run_index", run.RepoID)
-if err != nil {
-return err
-}
-run.Index = index
-run.Title = util.EllipsisDisplayString(run.Title, 255)
-
-if err := db.Insert(ctx, run); err != nil {
-return err
-}
-
-if run.Repo == nil {
-repo, err := repo_model.GetRepositoryByID(ctx, run.RepoID)
+return db.WithTx(ctx, func(ctx context.Context) error {
+index, err := db.GetNextResourceIndex(ctx, "action_run_index", run.RepoID)
 if err != nil {
 return err
 }
-run.Repo = repo
-}
+run.Index = index
+run.Title = util.EllipsisDisplayString(run.Title, 255)
 
-if err := updateRepoRunsNumbers(ctx, run.Repo); err != nil {
-return err
-}
-
-runJobs := make([]*ActionRunJob, 0, len(jobs))
-var hasWaiting bool
-for _, v := range jobs {
-id, job := v.Job()
-needs := job.Needs()
-if err := v.SetJob(id, job.EraseNeeds()); err != nil {
+if err := db.Insert(ctx, run); err != nil {
 return err
 }
-payload, _ := v.Marshal()
-status := StatusWaiting
-if len(needs) > 0 || run.NeedApproval {
-status = StatusBlocked
-} else {
-hasWaiting = true
+
+if run.Repo == nil {
+repo, err := repo_model.GetRepositoryByID(ctx, run.RepoID)
+if err != nil {
+return err
+}
+run.Repo = repo
 }
-job.Name = util.EllipsisDisplayString(job.Name, 255)
-runJobs = append(runJobs, &ActionRunJob{
-RunID: run.ID,
-RepoID: run.RepoID,
-OwnerID: run.OwnerID,
-CommitSHA: run.CommitSHA,
-IsForkPullRequest: run.IsForkPullRequest,
-Name: job.Name,
-WorkflowPayload: payload,
-JobID: id,
-Needs: needs,
-RunsOn: job.RunsOn(),
-Status: status,
-})
-}
-if err := db.Insert(ctx, runJobs); err != nil {
-return err
-}
 
-// if there is a job in the waiting status, increase tasks version.
-if hasWaiting {
-if err := IncreaseTaskVersion(ctx, run.OwnerID, run.RepoID); err != nil {
+if err := updateRepoRunsNumbers(ctx, run.Repo); err != nil {
 return err
 }
-}
 
-return committer.Commit()
+runJobs := make([]*ActionRunJob, 0, len(jobs))
+var hasWaiting bool
+for _, v := range jobs {
+id, job := v.Job()
+needs := job.Needs()
+if err := v.SetJob(id, job.EraseNeeds()); err != nil {
+return err
+}
+payload, _ := v.Marshal()
+status := StatusWaiting
+if len(needs) > 0 || run.NeedApproval {
+status = StatusBlocked
+} else {
+hasWaiting = true
+}
+job.Name = util.EllipsisDisplayString(job.Name, 255)
+runJobs = append(runJobs, &ActionRunJob{
+RunID: run.ID,
+RepoID: run.RepoID,
+OwnerID: run.OwnerID,
+CommitSHA: run.CommitSHA,
+IsForkPullRequest: run.IsForkPullRequest,
+Name: job.Name,
+WorkflowPayload: payload,
+JobID: id,
+Needs: needs,
+RunsOn: job.RunsOn(),
+Status: status,
+})
+}
+if err := db.Insert(ctx, runJobs); err != nil {
+return err
+}
+
+// if there is a job in the waiting status, increase tasks version.
+if hasWaiting {
+if err := IncreaseTaskVersion(ctx, run.OwnerID, run.RepoID); err != nil {
+return err
+}
+}
+return nil
+})
 }
 
 func GetRunByRepoAndID(ctx context.Context, repoID, runID int64) (*ActionRun, error) {

models/actions/schedule.go

@@ -56,65 +56,54 @@ func CreateScheduleTask(ctx context.Context, rows []*ActionSchedule) error {
 return nil
 }
 
-// Begin transaction
-ctx, committer, err := db.TxContext(ctx)
-if err != nil {
-return err
-}
-defer committer.Close()
-
-// Loop through each schedule row
-for _, row := range rows {
-row.Title = util.EllipsisDisplayString(row.Title, 255)
-// Create new schedule row
-if err = db.Insert(ctx, row); err != nil {
-return err
-}
-
-// Loop through each schedule spec and create a new spec row
-now := time.Now()
-
-for _, spec := range row.Specs {
-specRow := &ActionScheduleSpec{
-RepoID: row.RepoID,
-ScheduleID: row.ID,
-Spec: spec,
-}
-// Parse the spec and check for errors
-schedule, err := specRow.Parse()
-if err != nil {
-continue // skip to the next spec if there's an error
+return db.WithTx(ctx, func(ctx context.Context) error {
+// Loop through each schedule row
+for _, row := range rows {
+row.Title = util.EllipsisDisplayString(row.Title, 255)
+// Create new schedule row
+if err := db.Insert(ctx, row); err != nil {
+return err
 }
 
-specRow.Next = timeutil.TimeStamp(schedule.Next(now).Unix())
-
-// Insert the new schedule spec row
-if err = db.Insert(ctx, specRow); err != nil {
-return err
+// Loop through each schedule spec and create a new spec row
+now := time.Now()
+
+for _, spec := range row.Specs {
+specRow := &ActionScheduleSpec{
+RepoID: row.RepoID,
+ScheduleID: row.ID,
+Spec: spec,
+}
+// Parse the spec and check for errors
+schedule, err := specRow.Parse()
+if err != nil {
+continue // skip to the next spec if there's an error
+}
+
+specRow.Next = timeutil.TimeStamp(schedule.Next(now).Unix())
+
+// Insert the new schedule spec row
+if err = db.Insert(ctx, specRow); err != nil {
+return err
+}
 }
 }
-}
-
-// Commit transaction
-return committer.Commit()
+return nil
+})
 }
 
 func DeleteScheduleTaskByRepo(ctx context.Context, id int64) error {
-ctx, committer, err := db.TxContext(ctx)
-if err != nil {
-return err
-}
-defer committer.Close()
-
-if _, err := db.GetEngine(ctx).Delete(&ActionSchedule{RepoID: id}); err != nil {
-return err
-}
+return db.WithTx(ctx, func(ctx context.Context) error {
+if _, err := db.GetEngine(ctx).Delete(&ActionSchedule{RepoID: id}); err != nil {
+return err
+}
 
-if _, err := db.GetEngine(ctx).Delete(&ActionScheduleSpec{RepoID: id}); err != nil {
-return err
-}
+if _, err := db.GetEngine(ctx).Delete(&ActionScheduleSpec{RepoID: id}); err != nil {
+return err
+}
 
-return committer.Commit()
+return nil
+})
 }
 
 func CleanRepoScheduleTasks(ctx context.Context, repo *repo_model.Repository) ([]*ActionRunJob, error) {

models/actions/task.go

@@ -352,78 +352,70 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task
 stepStates[v.Id] = v
 }
 
-ctx, committer, err := db.TxContext(ctx)
-if err != nil {
-return nil, err
-}
-defer committer.Close()
-
-e := db.GetEngine(ctx)
-
-task := &ActionTask{}
-if has, err := e.ID(state.Id).Get(task); err != nil {
-return nil, err
-} else if !has {
-return nil, util.ErrNotExist
-} else if runnerID != task.RunnerID {
-return nil, errors.New("invalid runner for task")
-}
-
-if task.Status.IsDone() {
-// the state is final, do nothing
-return task, nil
-}
+return db.WithTx2(ctx, func(ctx context.Context) (*ActionTask, error) {
+e := db.GetEngine(ctx)
 
-// state.Result is not unspecified means the task is finished
-if state.Result != runnerv1.Result_RESULT_UNSPECIFIED {
-task.Status = Status(state.Result)
-task.Stopped = timeutil.TimeStamp(state.StoppedAt.AsTime().Unix())
-if err := UpdateTask(ctx, task, "status", "stopped"); err != nil {
+task := &ActionTask{}
+if has, err := e.ID(state.Id).Get(task); err != nil {
 return nil, err
+} else if !has {
+return nil, util.ErrNotExist
+} else if runnerID != task.RunnerID {
+return nil, errors.New("invalid runner for task")
 }
-if _, err := UpdateRunJob(ctx, &ActionRunJob{
-ID: task.JobID,
-Status: task.Status,
-Stopped: task.Stopped,
-}, nil); err != nil {
-return nil, 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
-}
-}
-
-if err := task.LoadAttributes(ctx); err != nil {
-return nil, err
-}
 
-for _, step := range task.Steps {
-var result runnerv1.Result
-if v, ok := stepStates[step.Index]; ok {
-result = v.Result
-step.LogIndex = v.LogIndex
-step.LogLength = v.LogLength
-step.Started = convertTimestamp(v.StartedAt)
-step.Stopped = convertTimestamp(v.StoppedAt)
+if task.Status.IsDone() {
+// the state is final, do nothing
+return task, nil
 }
-if result != runnerv1.Result_RESULT_UNSPECIFIED {
-step.Status = Status(result)
-} else if step.Started != 0 {
-step.Status = StatusRunning
+
+// state.Result is not unspecified means the task is finished
+if state.Result != runnerv1.Result_RESULT_UNSPECIFIED {
+task.Status = Status(state.Result)
+task.Stopped = timeutil.TimeStamp(state.StoppedAt.AsTime().Unix())
+if err := UpdateTask(ctx, task, "status", "stopped"); err != nil {
+return nil, err
+}
+if _, err := UpdateRunJob(ctx, &ActionRunJob{
+ID: task.JobID,
+Status: task.Status,
+Stopped: task.Stopped,
+}, nil); err != nil {
+return nil, 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
+}
 }
-if _, err := e.ID(step.ID).Update(step); err != nil {
+
+if err := task.LoadAttributes(ctx); err != nil {
 return nil, err
 }
-}
 
-if err := committer.Commit(); err != nil {
-return nil, err
-}
+for _, step := range task.Steps {
+var result runnerv1.Result
+if v, ok := stepStates[step.Index]; ok {
+result = v.Result
+step.LogIndex = v.LogIndex
+step.LogLength = v.LogLength
+step.Started = convertTimestamp(v.StartedAt)
+step.Stopped = convertTimestamp(v.StoppedAt)
+}
+if result != runnerv1.Result_RESULT_UNSPECIFIED {
+step.Status = Status(result)
+} else if step.Started != 0 {
+step.Status = StatusRunning
+}
+if _, err := e.ID(step.ID).Update(step); err != nil {
+return nil, err
+}
+}
 
-return task, nil
+return task, nil
+})
 }
 
 func StopTask(ctx context.Context, taskID int64, status Status) error {