Move all push update operations to a queue (#10133)

* Fix test

* Add no queue for test only

* improve code

* Auto watch whatever branch operation

* Fix lint

* Rename noqueue to immediate

* Remove old PushUpdate function

* Fix tests

Co-authored-by: zeripath <art27@cantab.net>
Co-authored-by: techknowlogick <techknowlogick@gitea.io>
This commit is contained in:
Lunny Xiao 2020-09-11 22:14:48 +08:00 committed by GitHub
parent 910947fbcd
commit dd1a651b58
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 552 additions and 585 deletions

View file

@ -6,14 +6,12 @@ package repofiles
import (
"bytes"
"container/list"
"fmt"
"path"
"strings"
"time"
"code.gitea.io/gitea/models"
"code.gitea.io/gitea/modules/cache"
"code.gitea.io/gitea/modules/charset"
"code.gitea.io/gitea/modules/git"
"code.gitea.io/gitea/modules/lfs"
@ -22,7 +20,6 @@ import (
"code.gitea.io/gitea/modules/setting"
"code.gitea.io/gitea/modules/storage"
"code.gitea.io/gitea/modules/structs"
pull_service "code.gitea.io/gitea/services/pull"
stdcharset "golang.org/x/net/html/charset"
"golang.org/x/text/transform"
@ -466,298 +463,3 @@ func CreateOrUpdateRepoFile(repo *models.Repository, doer *models.User, opts *Up
}
return file, nil
}
// PushUpdateOptions defines the push update options
type PushUpdateOptions struct {
PusherID int64
PusherName string
RepoUserName string
RepoName string
RefFullName string
OldCommitID string
NewCommitID string
}
// IsNewRef return true if it's a first-time push to a branch, tag or etc.
func (opts PushUpdateOptions) IsNewRef() bool {
return opts.OldCommitID == git.EmptySHA
}
// IsDelRef return true if it's a deletion to a branch or tag
func (opts PushUpdateOptions) IsDelRef() bool {
return opts.NewCommitID == git.EmptySHA
}
// IsUpdateRef return true if it's an update operation
func (opts PushUpdateOptions) IsUpdateRef() bool {
return !opts.IsNewRef() && !opts.IsDelRef()
}
// IsTag return true if it's an operation to a tag
func (opts PushUpdateOptions) IsTag() bool {
return strings.HasPrefix(opts.RefFullName, git.TagPrefix)
}
// IsNewTag return true if it's a creation to a tag
func (opts PushUpdateOptions) IsNewTag() bool {
return opts.IsTag() && opts.IsNewRef()
}
// IsDelTag return true if it's a deletion to a tag
func (opts PushUpdateOptions) IsDelTag() bool {
return opts.IsTag() && opts.IsDelRef()
}
// IsBranch return true if it's a push to branch
func (opts PushUpdateOptions) IsBranch() bool {
return strings.HasPrefix(opts.RefFullName, git.BranchPrefix)
}
// IsNewBranch return true if it's the first-time push to a branch
func (opts PushUpdateOptions) IsNewBranch() bool {
return opts.IsBranch() && opts.IsNewRef()
}
// IsUpdateBranch return true if it's not the first push to a branch
func (opts PushUpdateOptions) IsUpdateBranch() bool {
return opts.IsBranch() && opts.IsUpdateRef()
}
// IsDelBranch return true if it's a deletion to a branch
func (opts PushUpdateOptions) IsDelBranch() bool {
return opts.IsBranch() && opts.IsDelRef()
}
// TagName returns simple tag name if it's an operation to a tag
func (opts PushUpdateOptions) TagName() string {
return opts.RefFullName[len(git.TagPrefix):]
}
// BranchName returns simple branch name if it's an operation to branch
func (opts PushUpdateOptions) BranchName() string {
return opts.RefFullName[len(git.BranchPrefix):]
}
// RepoFullName returns repo full name
func (opts PushUpdateOptions) RepoFullName() string {
return opts.RepoUserName + "/" + opts.RepoName
}
// PushUpdate must be called for any push actions in order to
// generates necessary push action history feeds and other operations
func PushUpdate(repo *models.Repository, branch string, opts PushUpdateOptions) error {
if opts.IsNewRef() && opts.IsDelRef() {
return fmt.Errorf("Old and new revisions are both %s", git.EmptySHA)
}
repoPath := models.RepoPath(opts.RepoUserName, opts.RepoName)
_, err := git.NewCommand("update-server-info").RunInDir(repoPath)
if err != nil {
return fmt.Errorf("Failed to call 'git update-server-info': %v", err)
}
gitRepo, err := git.OpenRepository(repoPath)
if err != nil {
return fmt.Errorf("OpenRepository: %v", err)
}
defer gitRepo.Close()
if err = repo.UpdateSize(models.DefaultDBContext()); err != nil {
log.Error("Failed to update size for repository: %v", err)
}
var commits = &repo_module.PushCommits{}
if opts.IsTag() { // If is tag reference
tagName := opts.TagName()
if opts.IsDelRef() {
if err := models.PushUpdateDeleteTag(repo, tagName); err != nil {
return fmt.Errorf("PushUpdateDeleteTag: %v", err)
}
} else {
// Clear cache for tag commit count
cache.Remove(repo.GetCommitsCountCacheKey(tagName, true))
if err := repo_module.PushUpdateAddTag(repo, gitRepo, tagName); err != nil {
return fmt.Errorf("PushUpdateAddTag: %v", err)
}
}
} else if opts.IsBranch() { // If is branch reference
pusher, err := models.GetUserByID(opts.PusherID)
if err != nil {
return err
}
if !opts.IsDelRef() {
// Clear cache for branch commit count
cache.Remove(repo.GetCommitsCountCacheKey(opts.BranchName(), true))
newCommit, err := gitRepo.GetCommit(opts.NewCommitID)
if err != nil {
return fmt.Errorf("gitRepo.GetCommit: %v", err)
}
// Push new branch.
var l *list.List
if opts.IsNewRef() {
l, err = newCommit.CommitsBeforeLimit(10)
if err != nil {
return fmt.Errorf("newCommit.CommitsBeforeLimit: %v", err)
}
} else {
l, err = newCommit.CommitsBeforeUntil(opts.OldCommitID)
if err != nil {
return fmt.Errorf("newCommit.CommitsBeforeUntil: %v", err)
}
}
commits = repo_module.ListToPushCommits(l)
if err = models.RemoveDeletedBranch(repo.ID, opts.BranchName()); err != nil {
log.Error("models.RemoveDeletedBranch %s/%s failed: %v", repo.ID, opts.BranchName(), err)
}
if err = models.WatchIfAuto(opts.PusherID, repo.ID, true); err != nil {
log.Warn("Fail to perform auto watch on user %v for repo %v: %v", opts.PusherID, repo.ID, err)
}
log.Trace("TriggerTask '%s/%s' by %s", repo.Name, branch, pusher.Name)
go pull_service.AddTestPullRequestTask(pusher, repo.ID, branch, true, opts.OldCommitID, opts.NewCommitID)
} else if err = pull_service.CloseBranchPulls(pusher, repo.ID, branch); err != nil {
// close all related pulls
log.Error("close related pull request failed: %v", err)
}
}
if err := CommitRepoAction(&CommitRepoActionOptions{
PushUpdateOptions: opts,
RepoOwnerID: repo.OwnerID,
Commits: commits,
}); err != nil {
return fmt.Errorf("CommitRepoAction: %v", err)
}
return nil
}
// PushUpdates generates push action history feeds for push updating multiple refs
func PushUpdates(repo *models.Repository, optsList []*PushUpdateOptions) error {
repoPath := repo.RepoPath()
_, err := git.NewCommand("update-server-info").RunInDir(repoPath)
if err != nil {
return fmt.Errorf("Failed to call 'git update-server-info': %v", err)
}
gitRepo, err := git.OpenRepository(repoPath)
if err != nil {
return fmt.Errorf("OpenRepository: %v", err)
}
defer gitRepo.Close()
if err = repo.UpdateSize(models.DefaultDBContext()); err != nil {
log.Error("Failed to update size for repository: %v", err)
}
actions, err := createCommitRepoActions(repo, gitRepo, optsList)
if err != nil {
return err
}
if err := CommitRepoAction(actions...); err != nil {
return fmt.Errorf("CommitRepoAction: %v", err)
}
var pusher *models.User
for _, opts := range optsList {
if !opts.IsBranch() {
continue
}
branch := opts.BranchName()
if pusher == nil || pusher.ID != opts.PusherID {
var err error
pusher, err = models.GetUserByID(opts.PusherID)
if err != nil {
return err
}
}
if !opts.IsDelRef() {
if err = models.RemoveDeletedBranch(repo.ID, branch); err != nil {
log.Error("models.RemoveDeletedBranch %s/%s failed: %v", repo.ID, branch, err)
}
if err = models.WatchIfAuto(opts.PusherID, repo.ID, true); err != nil {
log.Warn("Fail to perform auto watch on user %v for repo %v: %v", opts.PusherID, repo.ID, err)
}
log.Trace("TriggerTask '%s/%s' by %s", repo.Name, branch, pusher.Name)
go pull_service.AddTestPullRequestTask(pusher, repo.ID, branch, true, opts.OldCommitID, opts.NewCommitID)
// close all related pulls
} else if err = pull_service.CloseBranchPulls(pusher, repo.ID, branch); err != nil {
log.Error("close related pull request failed: %v", err)
}
}
return nil
}
func createCommitRepoActions(repo *models.Repository, gitRepo *git.Repository, optsList []*PushUpdateOptions) ([]*CommitRepoActionOptions, error) {
addTags := make([]string, 0, len(optsList))
delTags := make([]string, 0, len(optsList))
actions := make([]*CommitRepoActionOptions, 0, len(optsList))
for _, opts := range optsList {
if opts.IsNewRef() && opts.IsDelRef() {
return nil, fmt.Errorf("Old and new revisions are both %s", git.EmptySHA)
}
var commits = &repo_module.PushCommits{}
if opts.IsTag() {
// If is tag reference
tagName := opts.TagName()
if opts.IsDelRef() {
delTags = append(delTags, tagName)
} else {
cache.Remove(repo.GetCommitsCountCacheKey(tagName, true))
addTags = append(addTags, tagName)
}
} else if !opts.IsDelRef() {
// If is branch reference
// Clear cache for branch commit count
cache.Remove(repo.GetCommitsCountCacheKey(opts.BranchName(), true))
newCommit, err := gitRepo.GetCommit(opts.NewCommitID)
if err != nil {
return nil, fmt.Errorf("gitRepo.GetCommit: %v", err)
}
// Push new branch.
var l *list.List
if opts.IsNewRef() {
l, err = newCommit.CommitsBeforeLimit(10)
if err != nil {
return nil, fmt.Errorf("newCommit.CommitsBeforeLimit: %v", err)
}
} else {
l, err = newCommit.CommitsBeforeUntil(opts.OldCommitID)
if err != nil {
return nil, fmt.Errorf("newCommit.CommitsBeforeUntil: %v", err)
}
}
commits = repo_module.ListToPushCommits(l)
}
actions = append(actions, &CommitRepoActionOptions{
PushUpdateOptions: *opts,
RepoOwnerID: repo.OwnerID,
Commits: commits,
})
}
if err := repo_module.PushUpdateAddDeleteTags(repo, gitRepo, addTags, delTags); err != nil {
return nil, fmt.Errorf("PushUpdateAddDeleteTags: %v", err)
}
return actions, nil
}