Revert "PullService lock via pullID (go-gitea#19520)" (for now...)

This reverts commit 6cde7c9159a5ea75a10356feb7b8c7ad4c434a9a.

Author: 6543

services/pull/check.go

@@ -28,8 +28,8 @@ import (
 	asymkey_service "code.gitea.io/gitea/services/asymkey"
 )
 
-// prPatchCheckerQueue represents a queue to handle update pull request tests
-var prPatchCheckerQueue queue.UniqueQueue
+// prQueue represents a queue to handle update pull request tests
+var prQueue queue.UniqueQueue
 
 var (
 	ErrIsClosed              = errors.New("pull is cosed")
@@ -43,7 +43,7 @@ var (
 
 // AddToTaskQueue adds itself to pull request test task queue.
 func AddToTaskQueue(pr *models.PullRequest) {
-	err := prPatchCheckerQueue.PushFunc(strconv.FormatInt(pr.ID, 10), func() error {
+	err := prQueue.PushFunc(strconv.FormatInt(pr.ID, 10), func() error {
 		pr.Status = models.PullRequestStatusChecking
 		err := pr.UpdateColsIfNotMerged("status")
 		if err != nil {
@@ -151,7 +151,7 @@ func checkAndUpdateStatus(pr *models.PullRequest) {
 	}
 
 	// Make sure there is no waiting test to process before leaving the checking status.
-	has, err := prPatchCheckerQueue.Has(strconv.FormatInt(pr.ID, 10))
+	has, err := prQueue.Has(strconv.FormatInt(pr.ID, 10))
 	if err != nil {
 		log.Error("Unable to check if the queue is waiting to reprocess pr.ID %d. Error: %v", pr.ID, err)
 	}
@@ -300,7 +300,7 @@ func InitializePullRequests(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		default:
-			if err := prPatchCheckerQueue.PushFunc(strconv.FormatInt(prID, 10), func() error {
+			if err := prQueue.PushFunc(strconv.FormatInt(prID, 10), func() error {
 				log.Trace("Adding PR ID: %d to the pull requests patch checking queue", prID)
 				return nil
 			}); err != nil {
@@ -321,8 +321,6 @@ func handle(data ...queue.Data) []queue.Data {
 }
 
 func testPR(id int64) {
-	pullWorkingPool.CheckIn(fmt.Sprint(id))
-	defer pullWorkingPool.CheckOut(fmt.Sprint(id))
 	ctx, _, finished := process.GetManager().AddContext(graceful.GetManager().HammerContext(), fmt.Sprintf("Test PR[%d] from patch checking queue", id))
 	defer finished()
 
@@ -367,13 +365,13 @@ func CheckPrsForBaseBranch(baseRepo *repo_model.Repository, baseBranchName strin
 
 // Init runs the task queue to test all the checking status pull requests
 func Init() error {
-	prPatchCheckerQueue = queue.CreateUniqueQueue("pr_patch_checker", handle, "")
+	prQueue = queue.CreateUniqueQueue("pr_patch_checker", handle, "")
 
-	if prPatchCheckerQueue == nil {
+	if prQueue == nil {
 		return fmt.Errorf("Unable to create pr_patch_checker Queue")
 	}
 
-	go graceful.GetManager().RunWithShutdownFns(prPatchCheckerQueue.Run)
+	go graceful.GetManager().RunWithShutdownFns(prQueue.Run)
 	go graceful.GetManager().RunWithShutdownContext(InitializePullRequests)
 	return nil
 }

services/pull/check_test.go

@@ -41,7 +41,7 @@ func TestPullRequest_AddToTaskQueue(t *testing.T) {
 	queueShutdown := []func(){}
 	queueTerminate := []func(){}
 
-	prPatchCheckerQueue = q.(queue.UniqueQueue)
+	prQueue = q.(queue.UniqueQueue)
 
 	pr := unittest.AssertExistsAndLoadBean(t, &models.PullRequest{ID: 2}).(*models.PullRequest)
 	AddToTaskQueue(pr)
@@ -51,11 +51,11 @@ func TestPullRequest_AddToTaskQueue(t *testing.T) {
 		return pr.Status == models.PullRequestStatusChecking
 	}, 1*time.Second, 100*time.Millisecond)
 
-	has, err := prPatchCheckerQueue.Has(strconv.FormatInt(pr.ID, 10))
+	has, err := prQueue.Has(strconv.FormatInt(pr.ID, 10))
 	assert.True(t, has)
 	assert.NoError(t, err)
 
-	prPatchCheckerQueue.Run(func(shutdown func()) {
+	prQueue.Run(func(shutdown func()) {
 		queueShutdown = append(queueShutdown, shutdown)
 	}, func(terminate func()) {
 		queueTerminate = append(queueTerminate, terminate)
@@ -68,7 +68,7 @@ func TestPullRequest_AddToTaskQueue(t *testing.T) {
 		assert.Fail(t, "Timeout: nothing was added to pullRequestQueue")
 	}
 
-	has, err = prPatchCheckerQueue.Has(strconv.FormatInt(pr.ID, 10))
+	has, err = prQueue.Has(strconv.FormatInt(pr.ID, 10))
 	assert.False(t, has)
 	assert.NoError(t, err)
 
@@ -82,5 +82,5 @@ func TestPullRequest_AddToTaskQueue(t *testing.T) {
 		callback()
 	}
 
-	prPatchCheckerQueue = nil
+	prQueue = nil
 }

services/pull/merge.go

@@ -34,6 +34,7 @@ import (
 
 // Merge merges pull request to base repository.
 // Caller should check PR is ready to be merged (review and status checks)
+// FIXME: add repoWorkingPull make sure two merges does not happen at same time.
 func Merge(ctx context.Context, pr *models.PullRequest, doer *user_model.User, baseGitRepo *git.Repository, mergeStyle repo_model.MergeStyle, expectedHeadCommitID, message string) (err error) {
 	if err = pr.LoadHeadRepoCtx(ctx); err != nil {
 		log.Error("LoadHeadRepo: %v", err)
@@ -43,9 +44,6 @@ func Merge(ctx context.Context, pr *models.PullRequest, doer *user_model.User, b
 		return fmt.Errorf("LoadBaseRepo: %v", err)
 	}
 
-	pullWorkingPool.CheckIn(fmt.Sprint(pr.ID))
-	defer pullWorkingPool.CheckOut(fmt.Sprint(pr.ID))
-
 	// Removing an auto merge pull and ignore if not exist
 	if err := pull_model.RemoveScheduledAutoMerge(ctx, doer, pr.ID, false); err != nil && !models.IsErrNotExist(err) {
 		return err
@@ -730,9 +728,6 @@ func CheckPRReadyToMerge(ctx context.Context, pr *models.PullRequest, skipProtec
 
 // MergedManually mark pr as merged manually
 func MergedManually(ctx context.Context, pr *models.PullRequest, doer *user_model.User, baseGitRepo *git.Repository, commitID string) (err error) {
-	pullWorkingPool.CheckIn(fmt.Sprint(pr.ID))
-	defer pullWorkingPool.CheckOut(fmt.Sprint(pr.ID))
-
 	prUnit, err := pr.BaseRepo.GetUnitCtx(ctx, unit.TypePullRequests)
 	if err != nil {
 		return

services/pull/pull.go

@@ -25,13 +25,9 @@ import (
 	"code.gitea.io/gitea/modules/notification"
 	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/setting"
-	"code.gitea.io/gitea/modules/sync"
 	issue_service "code.gitea.io/gitea/services/issue"
 )
 
-// TODO: use clustered lock (unique queue? or *abuse* cache)
-var pullWorkingPool = sync.NewExclusivePool()
-
 // NewPullRequest creates new pull request with labels for repository.
 func NewPullRequest(ctx context.Context, repo *repo_model.Repository, pull *models.Issue, labelIDs []int64, uuids []string, pr *models.PullRequest, assigneeIDs []int64) error {
 	if err := TestPatch(pr); err != nil {
@@ -128,9 +124,6 @@ func NewPullRequest(ctx context.Context, repo *repo_model.Repository, pull *mode
 
 // ChangeTargetBranch changes the target branch of this pull request, as the given user.
 func ChangeTargetBranch(ctx context.Context, pr *models.PullRequest, doer *user_model.User, targetBranch string) (err error) {
-	pullWorkingPool.CheckIn(fmt.Sprint(pr.ID))
-	defer pullWorkingPool.CheckOut(fmt.Sprint(pr.ID))
-
 	// Current target branch is already the same
 	if pr.BaseBranch == targetBranch {
 		return nil

services/pull/update.go

@@ -24,9 +24,6 @@ func Update(ctx context.Context, pull *models.PullRequest, doer *user_model.User
 		style repo_model.MergeStyle
 	)
 
-	pullWorkingPool.CheckIn(fmt.Sprint(pull.ID))
-	defer pullWorkingPool.CheckOut(fmt.Sprint(pull.ID))
-
 	if rebase {
 		pr = pull
 		style = repo_model.MergeStyleRebaseUpdate

services/repository/transfer.go

@@ -19,7 +19,6 @@ import (
 )
 
 // repoWorkingPool represents a working pool to order the parallel changes to the same repository
-// TODO: use clustered lock (unique queue? or *abuse* cache)
 var repoWorkingPool = sync.NewExclusivePool()
 
 // TransferOwnership transfers all corresponding setting from old user to new one.

services/wiki/wiki.go

@@ -27,8 +27,7 @@ import (
 
 var (
 	reservedWikiNames = []string{"_pages", "_new", "_edit", "raw"}
-	// TODO: use clustered lock (unique queue? or *abuse* cache)
-	wikiWorkingPool = sync.NewExclusivePool()
+	wikiWorkingPool   = sync.NewExclusivePool()
 )
 
 func nameAllowed(name string) error {