Refactor Cron and merge dashboard tasks (#10745)

* Refactor Cron and merge dashboard tasks

* Merge Cron and Dashboard tasks
* Make every cron task report a system notice on completion
* Refactor the creation of these tasks
* Ensure that execution counts of tasks is correct
* Allow cron tasks to be started from the cron page

* golangci-lint fixes

* Enforce that only one task with the same name can be registered

Signed-off-by: Andrew Thornton <art27@cantab.net>

* fix name check

Signed-off-by: Andrew Thornton <art27@cantab.net>

* as per @guillep2k

* as per @lafriks

Signed-off-by: Andrew Thornton <art27@cantab.net>

* Add git.CommandContext variants

Signed-off-by: Andrew Thornton <art27@cantab.net>

Co-authored-by: Lauris BH <lauris@nix.lv>
Co-authored-by: Lunny Xiao <xiaolunwen@gmail.com>
Co-authored-by: techknowlogick <techknowlogick@gitea.io>

5 files changed, 106 insertions, 52 deletions

diff --git a/models/admin.go b/models/admin.go
index 271c5307a0..01f0ba3486 100644
--- a/models/admin.go
+++ b/models/admin.go
@@ -20,6 +20,8 @@ type NoticeType int
 const (
 	//NoticeRepository type
 	NoticeRepository NoticeType = iota + 1
+	// NoticeTask type
+	NoticeTask
 )
 
 // Notice represents a system notice for admin.
@@ -36,11 +38,14 @@ func (n *Notice) TrStr() string {
 }
 
 // CreateNotice creates new system notice.
-func CreateNotice(tp NoticeType, desc string) error {
-	return createNotice(x, tp, desc)
+func CreateNotice(tp NoticeType, desc string, args ...interface{}) error {
+	return createNotice(x, tp, desc, args...)
 }
 
-func createNotice(e Engine, tp NoticeType, desc string) error {
+func createNotice(e Engine, tp NoticeType, desc string, args ...interface{}) error {
+	if len(args) > 0 {
+		desc = fmt.Sprintf(desc, args...)
+	}
 	n := &Notice{
 		Type:        tp,
 		Description: desc,
@@ -50,8 +55,8 @@ func createNotice(e Engine, tp NoticeType, desc string) error {
 }
 
 // CreateRepositoryNotice creates new system notice with type NoticeRepository.
-func CreateRepositoryNotice(desc string) error {
-	return createNotice(x, NoticeRepository, desc)
+func CreateRepositoryNotice(desc string, args ...interface{}) error {
+	return createNotice(x, NoticeRepository, desc, args...)
 }
 
 // RemoveAllWithNotice removes all directories in given path and
diff --git a/models/branches.go b/models/branches.go
index e6b8d61a70..688e2d7f51 100644
--- a/models/branches.go
+++ b/models/branches.go
@@ -12,7 +12,6 @@ import (
 
 	"code.gitea.io/gitea/modules/base"
 	"code.gitea.io/gitea/modules/log"
-	"code.gitea.io/gitea/modules/setting"
 	"code.gitea.io/gitea/modules/timeutil"
 	"code.gitea.io/gitea/modules/util"
 
@@ -561,11 +560,11 @@ func RemoveDeletedBranch(repoID int64, branch string) error {
 }
 
 // RemoveOldDeletedBranches removes old deleted branches
-func RemoveOldDeletedBranches(ctx context.Context) {
+func RemoveOldDeletedBranches(ctx context.Context, olderThan time.Duration) {
 	// Nothing to do for shutdown or terminate
 	log.Trace("Doing: DeletedBranchesCleanup")
 
-	deleteBefore := time.Now().Add(-setting.Cron.DeletedBranchesCleanup.OlderThan)
+	deleteBefore := time.Now().Add(-olderThan)
 	_, err := x.Where("deleted_unix < ?", deleteBefore.Unix()).Delete(new(DeletedBranch))
 	if err != nil {
 		log.Error("DeletedBranchesCleanup: %v", err)
diff --git a/models/error.go b/models/error.go
index 7370bd1571..3b05a7152c 100644
--- a/models/error.go
+++ b/models/error.go
@@ -85,6 +85,28 @@ func (err ErrSSHDisabled) Error() string {
 	return "SSH is disabled"
 }
 
+// ErrCancelled represents an error due to context cancellation
+type ErrCancelled struct {
+	Message string
+}
+
+// IsErrCancelled checks if an error is a ErrCancelled.
+func IsErrCancelled(err error) bool {
+	_, ok := err.(ErrCancelled)
+	return ok
+}
+
+func (err ErrCancelled) Error() string {
+	return "Cancelled: " + err.Message
+}
+
+// ErrCancelledf returns an ErrCancelled for the provided format and args
+func ErrCancelledf(format string, args ...interface{}) error {
+	return ErrCancelled{
+		fmt.Sprintf(format, args...),
+	}
+}
+
 //  ____ ___
 // |    |   \______ ___________
 // |    |   /  ___// __ \_  __ \
diff --git a/models/repo.go b/models/repo.go
index 875c9ad1c3..f79740e747 100644
--- a/models/repo.go
+++ b/models/repo.go
@@ -1853,35 +1853,44 @@ func GetPrivateRepositoryCount(u *User) (int64, error) {
 }
 
 // DeleteRepositoryArchives deletes all repositories' archives.
-func DeleteRepositoryArchives() error {
+func DeleteRepositoryArchives(ctx context.Context) error {
 	return x.
 		Where("id > 0").
 		Iterate(new(Repository),
 			func(idx int, bean interface{}) error {
 				repo := bean.(*Repository)
+				select {
+				case <-ctx.Done():
+					return ErrCancelledf("before deleting repository archives for %s", repo.FullName())
+				default:
+				}
 				return os.RemoveAll(filepath.Join(repo.RepoPath(), "archives"))
 			})
 }
 
 // DeleteOldRepositoryArchives deletes old repository archives.
-func DeleteOldRepositoryArchives(ctx context.Context) {
+func DeleteOldRepositoryArchives(ctx context.Context, olderThan time.Duration) error {
 	log.Trace("Doing: ArchiveCleanup")
 
 	if err := x.Where("id > 0").Iterate(new(Repository), func(idx int, bean interface{}) error {
-		return deleteOldRepositoryArchives(ctx, idx, bean)
+		return deleteOldRepositoryArchives(ctx, olderThan, idx, bean)
 	}); err != nil {
-		log.Error("ArchiveClean: %v", err)
+		log.Trace("Error: ArchiveClean: %v", err)
+		return err
 	}
+
+	log.Trace("Finished: ArchiveCleanup")
+	return nil
 }
 
-func deleteOldRepositoryArchives(ctx context.Context, idx int, bean interface{}) error {
+func deleteOldRepositoryArchives(ctx context.Context, olderThan time.Duration, idx int, bean interface{}) error {
 	repo := bean.(*Repository)
 	basePath := filepath.Join(repo.RepoPath(), "archives")
 
 	for _, ty := range []string{"zip", "targz"} {
 		select {
 		case <-ctx.Done():
-			return fmt.Errorf("Aborted due to shutdown:\nin delete of old repository archives %v\nat delete file %s", repo, ty)
+			return ErrCancelledf("before deleting old repository archives with filetype %s for %s", ty, repo.FullName())
 		default:
 		}
 
@@ -1904,12 +1913,12 @@ func deleteOldRepositoryArchives(ctx context.Context, idx int, bean interface{})
 			return err
 		}
 
-		minimumOldestTime := time.Now().Add(-setting.Cron.ArchiveCleanup.OlderThan)
+		minimumOldestTime := time.Now().Add(-olderThan)
 		for _, info := range files {
 			if info.ModTime().Before(minimumOldestTime) && !info.IsDir() {
 				select {
 				case <-ctx.Done():
-					return fmt.Errorf("Aborted due to shutdown:\nin delete of old repository archives %v\nat delete file %s - %s", repo, ty, info.Name())
+					return ErrCancelledf("before deleting old repository archive file %s with filetype %s for %s", info.Name(), ty, repo.FullName())
 				default:
 				}
 				toDelete := filepath.Join(path, info.Name())
@@ -1936,13 +1945,13 @@ func repoStatsCheck(ctx context.Context, checker *repoChecker) {
 		return
 	}
 	for _, result := range results {
+		id := com.StrTo(result["id"]).MustInt64()
 		select {
 		case <-ctx.Done():
-			log.Warn("CheckRepoStats: Aborting due to shutdown")
+			log.Warn("CheckRepoStats: Cancelled before checking %s for Repo[%d]", checker.desc, id)
 			return
 		default:
 		}
-		id := com.StrTo(result["id"]).MustInt64()
 		log.Trace("Updating %s: %d", checker.desc, id)
 		_, err = x.Exec(checker.correctSQL, id, id)
 		if err != nil {
@@ -1952,7 +1961,7 @@ func repoStatsCheck(ctx context.Context, checker *repoChecker) {
 }
 
 // CheckRepoStats checks the repository stats
-func CheckRepoStats(ctx context.Context) {
+func CheckRepoStats(ctx context.Context) error {
 	log.Trace("Doing: CheckRepoStats")
 
 	checkers := []*repoChecker{
@@ -1987,13 +1996,13 @@ func CheckRepoStats(ctx context.Context) {
 			"issue count 'num_comments'",
 		},
 	}
-	for i := range checkers {
+	for _, checker := range checkers {
 		select {
 		case <-ctx.Done():
-			log.Warn("CheckRepoStats: Aborting due to shutdown")
-			return
+			log.Warn("CheckRepoStats: Cancelled before %s", checker.desc)
+			return ErrCancelledf("before checking %s", checker.desc)
 		default:
-			repoStatsCheck(ctx, checkers[i])
+			repoStatsCheck(ctx, checker)
 		}
 	}
 
@@ -2004,13 +2013,13 @@ func CheckRepoStats(ctx context.Context) {
 		log.Error("Select %s: %v", desc, err)
 	} else {
 		for _, result := range results {
+			id := com.StrTo(result["id"]).MustInt64()
 			select {
 			case <-ctx.Done():
-				log.Warn("CheckRepoStats: Aborting due to shutdown")
-				return
+				log.Warn("CheckRepoStats: Cancelled during %s for repo ID %d", desc, id)
+				return ErrCancelledf("during %s for repo ID %d", desc, id)
 			default:
 			}
-			id := com.StrTo(result["id"]).MustInt64()
 			log.Trace("Updating %s: %d", desc, id)
 			_, err = x.Exec("UPDATE `repository` SET num_closed_issues=(SELECT COUNT(*) FROM `issue` WHERE repo_id=? AND is_closed=? AND is_pull=?) WHERE id=?", id, true, false, id)
 			if err != nil {
@@ -2027,13 +2036,13 @@ func CheckRepoStats(ctx context.Context) {
 		log.Error("Select %s: %v", desc, err)
 	} else {
 		for _, result := range results {
+			id := com.StrTo(result["id"]).MustInt64()
 			select {
 			case <-ctx.Done():
-				log.Warn("CheckRepoStats: Aborting due to shutdown")
-				return
+				log.Warn("CheckRepoStats: Cancelled")
+				return ErrCancelledf("during %s for repo ID %d", desc, id)
 			default:
 			}
-			id := com.StrTo(result["id"]).MustInt64()
 			log.Trace("Updating %s: %d", desc, id)
 			_, err = x.Exec("UPDATE `repository` SET num_closed_pulls=(SELECT COUNT(*) FROM `issue` WHERE repo_id=? AND is_closed=? AND is_pull=?) WHERE id=?", id, true, true, id)
 			if err != nil {
@@ -2050,13 +2059,13 @@ func CheckRepoStats(ctx context.Context) {
 		log.Error("Select repository count 'num_forks': %v", err)
 	} else {
 		for _, result := range results {
+			id := com.StrTo(result["id"]).MustInt64()
 			select {
 			case <-ctx.Done():
-				log.Warn("CheckRepoStats: Aborting due to shutdown")
-				return
+				log.Warn("CheckRepoStats: Cancelled")
+				return ErrCancelledf("during %s for repo ID %d", desc, id)
 			default:
 			}
-			id := com.StrTo(result["id"]).MustInt64()
 			log.Trace("Updating repository count 'num_forks': %d", id)
 
 			repo, err := GetRepositoryByID(id)
@@ -2079,6 +2088,7 @@ func CheckRepoStats(ctx context.Context) {
 		}
 	}
 	// ***** END: Repository.NumForks *****
+	return nil
 }
 
 // SetArchiveRepoState sets if a repo is archived
@@ -2189,12 +2199,17 @@ func (repo *Repository) generateRandomAvatar(e Engine) error {
 }
 
 // RemoveRandomAvatars removes the randomly generated avatars that were created for repositories
-func RemoveRandomAvatars() error {
+func RemoveRandomAvatars(ctx context.Context) error {
 	return x.
 		Where("id > 0").BufferSize(setting.Database.IterateBufferSize).
 		Iterate(new(Repository),
 			func(idx int, bean interface{}) error {
 				repository := bean.(*Repository)
+				select {
+				case <-ctx.Done():
+					return ErrCancelledf("before random avatars removed for %s", repository.FullName())
+				default:
+				}
 				stringifiedID := strconv.FormatInt(repository.ID, 10)
 				if repository.Avatar == stringifiedID {
 					return repository.DeleteAvatar()
diff --git a/models/user.go b/models/user.go
index 01cce2506d..8875840db7 100644
--- a/models/user.go
+++ b/models/user.go
@@ -1321,16 +1321,30 @@ func DeleteUser(u *User) (err error) {
 	return sess.Commit()
 }
 
-// DeleteInactivateUsers deletes all inactivate users and email addresses.
-func DeleteInactivateUsers() (err error) {
+// DeleteInactiveUsers deletes all inactive users and email addresses.
+func DeleteInactiveUsers(ctx context.Context, olderThan time.Duration) (err error) {
 	users := make([]*User, 0, 10)
-	if err = x.
-		Where("is_active = ?", false).
-		Find(&users); err != nil {
-		return fmt.Errorf("get all inactive users: %v", err)
+	if olderThan > 0 {
+		if err = x.
+			Where("is_active = ? and created_unix < ?", false, time.Now().Add(-olderThan).Unix()).
+			Find(&users); err != nil {
+			return fmt.Errorf("get all inactive users: %v", err)
+		}
+	} else {
+		if err = x.
+			Where("is_active = ?", false).
+			Find(&users); err != nil {
+			return fmt.Errorf("get all inactive users: %v", err)
+		}
+
 	}
 	// FIXME: should only update authorized_keys file once after all deletions.
 	for _, u := range users {
+		select {
+		case <-ctx.Done():
+			return ErrCancelledf("Before delete inactive user %s", u.Name)
+		default:
+		}
 		if err = DeleteUser(u); err != nil {
 			// Ignore users that were set inactive by admin.
 			if IsErrUserOwnRepos(err) || IsErrUserHasOrgs(err) {
@@ -1814,25 +1828,23 @@ func synchronizeLdapSSHPublicKeys(usr *User, s *LoginSource, sshPublicKeys []str
 }
 
 // SyncExternalUsers is used to synchronize users with external authorization source
-func SyncExternalUsers(ctx context.Context) {
+func SyncExternalUsers(ctx context.Context, updateExisting bool) error {
 	log.Trace("Doing: SyncExternalUsers")
 
 	ls, err := LoginSources()
 	if err != nil {
 		log.Error("SyncExternalUsers: %v", err)
-		return
+		return err
 	}
 
-	updateExisting := setting.Cron.SyncExternalUsers.UpdateExisting
-
 	for _, s := range ls {
 		if !s.IsActived || !s.IsSyncEnabled {
 			continue
 		}
 		select {
 		case <-ctx.Done():
-			log.Warn("SyncExternalUsers: Aborted due to shutdown before update of %s", s.Name)
-			return
+			log.Warn("SyncExternalUsers: Cancelled before update of %s", s.Name)
+			return ErrCancelledf("Before update of %s", s.Name)
 		default:
 		}
 
@@ -1850,12 +1862,12 @@ func SyncExternalUsers(ctx context.Context) {
 				Find(&users)
 			if err != nil {
 				log.Error("SyncExternalUsers: %v", err)
-				return
+				return err
 			}
 			select {
 			case <-ctx.Done():
-				log.Warn("SyncExternalUsers: Aborted due to shutdown before update of %s", s.Name)
-				return
+				log.Warn("SyncExternalUsers: Cancelled before update of %s", s.Name)
+				return ErrCancelledf("Before update of %s", s.Name)
 			default:
 			}
 
@@ -1877,7 +1889,7 @@ func SyncExternalUsers(ctx context.Context) {
 			for _, su := range sr {
 				select {
 				case <-ctx.Done():
-					log.Warn("SyncExternalUsers: Aborted due to shutdown at update of %s before completed update of users", s.Name)
+					log.Warn("SyncExternalUsers: Cancelled at update of %s before completed update of users", s.Name)
 					// Rewrite authorized_keys file if LDAP Public SSH Key attribute is set and any key was added or removed
 					if sshKeysNeedUpdate {
 						err = RewriteAllPublicKeys()
@@ -1885,7 +1897,7 @@ func SyncExternalUsers(ctx context.Context) {
 							log.Error("RewriteAllPublicKeys: %v", err)
 						}
 					}
-					return
+					return ErrCancelledf("During update of %s before completed update of users", s.Name)
 				default:
 				}
 				if len(su.Username) == 0 {
@@ -1980,8 +1992,8 @@ func SyncExternalUsers(ctx context.Context) {
 
 			select {
 			case <-ctx.Done():
-				log.Warn("SyncExternalUsers: Aborted due to shutdown at update of %s before delete users", s.Name)
-				return
+				log.Warn("SyncExternalUsers: Cancelled during update of %s before delete users", s.Name)
+				return ErrCancelledf("During update of %s before delete users", s.Name)
 			default:
 			}
 
@@ -2008,4 +2020,5 @@ func SyncExternalUsers(ctx context.Context) {
 			}
 		}
 	}
+	return nil
 }