Fix schedule tasks bugs (#28691)

Fix #28157 

This PR fix the possible bugs about actions schedule.

## The Changes

- Move `UpdateRepositoryUnit` and `SetRepoDefaultBranch` from models to
service layer
- Remove schedules plan from database and cancel waiting & running
schedules tasks in this repository when actions unit has been disabled
or global disabled.
- Remove schedules plan from database and cancel waiting & running
schedules tasks in this repository when default branch changed.

services/actions/notifier_helper.go

@@ -117,6 +117,9 @@ func notify(ctx context.Context, input *notifyInput) error {
 		return nil
 	}
 	if unit_model.TypeActions.UnitGlobalDisabled() {
+		if err := actions_model.CleanRepoScheduleTasks(ctx, input.Repo); err != nil {
+			log.Error("CleanRepoScheduleTasks: %v", err)
+		}
 		return nil
 	}
 	if err := input.Repo.LoadUnits(ctx); err != nil {
@@ -153,7 +156,11 @@ func notify(ctx context.Context, input *notifyInput) error {
 
 	var detectedWorkflows []*actions_module.DetectedWorkflow
 	actionsConfig := input.Repo.MustGetUnit(ctx, unit_model.TypeActions).ActionsConfig()
-	workflows, schedules, err := actions_module.DetectWorkflows(gitRepo, commit, input.Event, input.Payload)
+	workflows, schedules, err := actions_module.DetectWorkflows(gitRepo, commit,
+		input.Event,
+		input.Payload,
+		input.Event == webhook_module.HookEventPush && input.Ref == input.Repo.DefaultBranch,
+	)
 	if err != nil {
 		return fmt.Errorf("DetectWorkflows: %w", err)
 	}
@@ -167,7 +174,7 @@ func notify(ctx context.Context, input *notifyInput) error {
 				continue
 			}
 
-			if wf.TriggerEvent != actions_module.GithubEventPullRequestTarget {
+			if wf.TriggerEvent.Name != actions_module.GithubEventPullRequestTarget {
 				detectedWorkflows = append(detectedWorkflows, wf)
 			}
 		}
@@ -180,7 +187,7 @@ func notify(ctx context.Context, input *notifyInput) error {
 		if err != nil {
 			return fmt.Errorf("gitRepo.GetCommit: %w", err)
 		}
-		baseWorkflows, _, err := actions_module.DetectWorkflows(gitRepo, baseCommit, input.Event, input.Payload)
+		baseWorkflows, _, err := actions_module.DetectWorkflows(gitRepo, baseCommit, input.Event, input.Payload, false)
 		if err != nil {
 			return fmt.Errorf("DetectWorkflows: %w", err)
 		}
@@ -188,7 +195,7 @@ func notify(ctx context.Context, input *notifyInput) error {
 			log.Trace("repo %s with commit %s couldn't find pull_request_target workflows", input.Repo.RepoPath(), baseCommit.ID)
 		} else {
 			for _, wf := range baseWorkflows {
-				if wf.TriggerEvent == actions_module.GithubEventPullRequestTarget {
+				if wf.TriggerEvent.Name == actions_module.GithubEventPullRequestTarget {
 					detectedWorkflows = append(detectedWorkflows, wf)
 				}
 			}
@@ -265,7 +272,7 @@ func handleWorkflows(
 			IsForkPullRequest: isForkPullRequest,
 			Event:             input.Event,
 			EventPayload:      string(p),
-			TriggerEvent:      dwf.TriggerEvent,
+			TriggerEvent:      dwf.TriggerEvent.Name,
 			Status:            actions_model.StatusWaiting,
 		}
 		if need, err := ifNeedApproval(ctx, run, input.Repo, input.Doer); err != nil {
@@ -289,6 +296,7 @@ func handleWorkflows(
 				run.RepoID,
 				run.Ref,
 				run.WorkflowID,
+				run.Event,
 			); err != nil {
 				log.Error("CancelRunningJobs: %v", err)
 			}
@@ -414,8 +422,8 @@ func handleSchedules(
 		log.Error("CountSchedules: %v", err)
 		return err
 	} else if count > 0 {
-		if err := actions_model.DeleteScheduleTaskByRepo(ctx, input.Repo.ID); err != nil {
-			log.Error("DeleteCronTaskByRepo: %v", err)
+		if err := actions_model.CleanRepoScheduleTasks(ctx, input.Repo); err != nil {
+			log.Error("CleanRepoScheduleTasks: %v", err)
 		}
 	}
 
@@ -456,19 +464,6 @@ func handleSchedules(
 			Specs:         schedules,
 			Content:       dwf.Content,
 		}
-
-		// cancel running jobs if the event is push
-		if run.Event == webhook_module.HookEventPush {
-			// cancel running jobs of the same workflow
-			if err := actions_model.CancelRunningJobs(
-				ctx,
-				run.RepoID,
-				run.Ref,
-				run.WorkflowID,
-			); err != nil {
-				log.Error("CancelRunningJobs: %v", err)
-			}
-		}
 		crons = append(crons, run)
 	}
 

services/actions/schedule_tasks.go

@@ -59,6 +59,7 @@ func startTasks(ctx context.Context) error {
 					row.RepoID,
 					row.Schedule.Ref,
 					row.Schedule.WorkflowID,
+					webhook_module.HookEventSchedule,
 				); err != nil {
 					log.Error("CancelRunningJobs: %v", err)
 				}
@@ -113,6 +114,7 @@ func CreateScheduleTask(ctx context.Context, cron *actions_model.ActionSchedule)
 		CommitSHA:     cron.CommitSHA,
 		Event:         cron.Event,
 		EventPayload:  cron.EventPayload,
+		TriggerEvent:  string(webhook_module.HookEventSchedule),
 		ScheduleID:    cron.ID,
 		Status:        actions_model.StatusWaiting,
 	}

services/repository/branch.go

@@ -10,6 +10,7 @@ import (
 	"strings"
 
 	"code.gitea.io/gitea/models"
+	actions_model "code.gitea.io/gitea/models/actions"
 	"code.gitea.io/gitea/models/db"
 	git_model "code.gitea.io/gitea/models/git"
 	issues_model "code.gitea.io/gitea/models/issues"
@@ -22,6 +23,7 @@ import (
 	repo_module "code.gitea.io/gitea/modules/repository"
 	"code.gitea.io/gitea/modules/timeutil"
 	"code.gitea.io/gitea/modules/util"
+	webhook_module "code.gitea.io/gitea/modules/webhook"
 	notify_service "code.gitea.io/gitea/services/notify"
 	files_service "code.gitea.io/gitea/services/repository/files"
 
@@ -308,13 +310,28 @@ func RenameBranch(ctx context.Context, repo *repo_model.Repository, doer *user_m
 		return "from_not_exist", nil
 	}
 
-	if err := git_model.RenameBranch(ctx, repo, from, to, func(isDefault bool) error {
+	if err := git_model.RenameBranch(ctx, repo, from, to, func(ctx context.Context, isDefault bool) error {
 		err2 := gitRepo.RenameBranch(from, to)
 		if err2 != nil {
 			return err2
 		}
 
 		if isDefault {
+			// if default branch changed, we need to delete all schedules and cron jobs
+			if err := actions_model.DeleteScheduleTaskByRepo(ctx, repo.ID); err != nil {
+				log.Error("DeleteCronTaskByRepo: %v", err)
+			}
+			// cancel running cron jobs of this repository and delete old schedules
+			if err := actions_model.CancelRunningJobs(
+				ctx,
+				repo.ID,
+				from,
+				"",
+				webhook_module.HookEventSchedule,
+			); err != nil {
+				log.Error("CancelRunningJobs: %v", err)
+			}
+
 			err2 = gitRepo.SetDefaultBranch(to)
 			if err2 != nil {
 				return err2
@@ -450,3 +467,50 @@ func AddAllRepoBranchesToSyncQueue(ctx context.Context, doerID int64) error {
 	}
 	return nil
 }
+
+func SetRepoDefaultBranch(ctx context.Context, repo *repo_model.Repository, gitRepo *git.Repository, newBranchName string) error {
+	if repo.DefaultBranch == newBranchName {
+		return nil
+	}
+
+	if !gitRepo.IsBranchExist(newBranchName) {
+		return git_model.ErrBranchNotExist{
+			BranchName: newBranchName,
+		}
+	}
+
+	oldDefaultBranchName := repo.DefaultBranch
+	repo.DefaultBranch = newBranchName
+	if err := db.WithTx(ctx, func(ctx context.Context) error {
+		if err := repo_model.UpdateDefaultBranch(ctx, repo); err != nil {
+			return err
+		}
+
+		if err := actions_model.DeleteScheduleTaskByRepo(ctx, repo.ID); err != nil {
+			log.Error("DeleteCronTaskByRepo: %v", err)
+		}
+		// cancel running cron jobs of this repository and delete old schedules
+		if err := actions_model.CancelRunningJobs(
+			ctx,
+			repo.ID,
+			oldDefaultBranchName,
+			"",
+			webhook_module.HookEventSchedule,
+		); err != nil {
+			log.Error("CancelRunningJobs: %v", err)
+		}
+
+		if err := gitRepo.SetDefaultBranch(newBranchName); err != nil {
+			if !git.IsErrUnsupportedVersion(err) {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	notify_service.ChangeDefaultBranch(ctx, repo)
+
+	return nil
+}

services/repository/setting.go

@@ -0,0 +1,47 @@
+// Copyright 2023 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package repository
+
+import (
+	"context"
+	"slices"
+
+	actions_model "code.gitea.io/gitea/models/actions"
+	"code.gitea.io/gitea/models/db"
+	repo_model "code.gitea.io/gitea/models/repo"
+	"code.gitea.io/gitea/models/unit"
+	"code.gitea.io/gitea/modules/log"
+)
+
+// UpdateRepositoryUnits updates a repository's units
+func UpdateRepositoryUnits(ctx context.Context, repo *repo_model.Repository, units []repo_model.RepoUnit, deleteUnitTypes []unit.Type) (err error) {
+	ctx, committer, err := db.TxContext(ctx)
+	if err != nil {
+		return err
+	}
+	defer committer.Close()
+
+	// Delete existing settings of units before adding again
+	for _, u := range units {
+		deleteUnitTypes = append(deleteUnitTypes, u.Type)
+	}
+
+	if slices.Contains(deleteUnitTypes, unit.TypeActions) {
+		if err := actions_model.CleanRepoScheduleTasks(ctx, repo); err != nil {
+			log.Error("CleanRepoScheduleTasks: %v", err)
+		}
+	}
+
+	if _, err = db.GetEngine(ctx).Where("repo_id = ?", repo.ID).In("type", deleteUnitTypes).Delete(new(repo_model.RepoUnit)); err != nil {
+		return err
+	}
+
+	if len(units) > 0 {
+		if err = db.Insert(ctx, units); err != nil {
+			return err
+		}
+	}
+
+	return committer.Commit()
+}

services/wiki/wiki.go

@@ -19,6 +19,7 @@ import (
 	repo_module "code.gitea.io/gitea/modules/repository"
 	"code.gitea.io/gitea/modules/sync"
 	asymkey_service "code.gitea.io/gitea/services/asymkey"
+	repo_service "code.gitea.io/gitea/services/repository"
 )
 
 // TODO: use clustered lock (unique queue? or *abuse* cache)
@@ -350,7 +351,7 @@ func DeleteWikiPage(ctx context.Context, doer *user_model.User, repo *repo_model
 
 // DeleteWiki removes the actual and local copy of repository wiki.
 func DeleteWiki(ctx context.Context, repo *repo_model.Repository) error {
-	if err := repo_model.UpdateRepositoryUnits(ctx, repo, nil, []unit.Type{unit.TypeWiki}); err != nil {
+	if err := repo_service.UpdateRepositoryUnits(ctx, repo, nil, []unit.Type{unit.TypeWiki}); err != nil {
 		return err
 	}