Fix parallelly generating index failure with Mysql (#24567) (#25081)

Backport #24567 by @lunny

- Fix possible parallel creating commit status index problem and
creating issues/pull request index problem work with Mysql5/Mysql8
- Add parallel tests
- Reenable TestRepoCommitsStatusParallel on CI

Fix #22109

---------

Co-authored-by: Lunny Xiao <xiaolunwen@gmail.com>

models/db/index.go

@@ -71,10 +71,31 @@ func postgresGetNextResourceIndex(ctx context.Context, tableName string, groupID
 	return strconv.ParseInt(string(res[0]["max_index"]), 10, 64)
 }
 
+func mysqlGetNextResourceIndex(ctx context.Context, tableName string, groupID int64) (int64, error) {
+	if _, err := GetEngine(ctx).Exec(fmt.Sprintf("INSERT INTO %s (group_id, max_index) "+
+		"VALUES (?,1) ON DUPLICATE KEY UPDATE max_index = max_index+1",
+		tableName), groupID); err != nil {
+		return 0, err
+	}
+
+	var idx int64
+	_, err := GetEngine(ctx).SQL(fmt.Sprintf("SELECT max_index FROM %s WHERE group_id = ?", tableName), groupID).Get(&idx)
+	if err != nil {
+		return 0, err
+	}
+	if idx == 0 {
+		return 0, errors.New("cannot get the correct index")
+	}
+	return idx, nil
+}
+
 // GetNextResourceIndex generates a resource index, it must run in the same transaction where the resource is created
 func GetNextResourceIndex(ctx context.Context, tableName string, groupID int64) (int64, error) {
-	if setting.Database.Type.IsPostgreSQL() {
+	switch {
+	case setting.Database.Type.IsPostgreSQL():
 		return postgresGetNextResourceIndex(ctx, tableName, groupID)
+	case setting.Database.Type.IsMySQL():
+		return mysqlGetNextResourceIndex(ctx, tableName, groupID)
 	}
 
 	e := GetEngine(ctx)

models/git/commit_status.go

@@ -63,18 +63,40 @@ func postgresGetCommitStatusIndex(ctx context.Context, repoID int64, sha string)
 	return strconv.ParseInt(string(res[0]["max_index"]), 10, 64)
 }
 
+func mysqlGetCommitStatusIndex(ctx context.Context, repoID int64, sha string) (int64, error) {
+	if _, err := db.GetEngine(ctx).Exec("INSERT INTO `commit_status_index` (repo_id, sha, max_index) "+
+		"VALUES (?,?,1) ON DUPLICATE KEY UPDATE max_index = max_index+1",
+		repoID, sha); err != nil {
+		return 0, err
+	}
+
+	var idx int64
+	_, err := db.GetEngine(ctx).SQL("SELECT max_index FROM `commit_status_index` WHERE repo_id = ? AND sha = ?",
+		repoID, sha).Get(&idx)
+	if err != nil {
+		return 0, err
+	}
+	if idx == 0 {
+		return 0, errors.New("cannot get the correct index")
+	}
+	return idx, nil
+}
+
 // GetNextCommitStatusIndex retried 3 times to generate a resource index
 func GetNextCommitStatusIndex(ctx context.Context, repoID int64, sha string) (int64, error) {
-	if setting.Database.Type.IsPostgreSQL() {
+	switch {
+	case setting.Database.Type.IsPostgreSQL():
 		return postgresGetCommitStatusIndex(ctx, repoID, sha)
+	case setting.Database.Type.IsMySQL():
+		return mysqlGetCommitStatusIndex(ctx, repoID, sha)
 	}
 
 	e := db.GetEngine(ctx)
 
 	// try to update the max_index to next value, and acquire the write-lock for the record
 	res, err := e.Exec("UPDATE `commit_status_index` SET max_index=max_index+1 WHERE repo_id=? AND sha=?", repoID, sha)
 	if err != nil {
-		return 0, err
+		return 0, fmt.Errorf("update failed: %w", err)
 	}
 	affected, err := res.RowsAffected()
 	if err != nil {
@@ -85,26 +107,26 @@ func GetNextCommitStatusIndex(ctx context.Context, repoID int64, sha string) (in
 		_, errIns := e.Exec("INSERT INTO `commit_status_index` (repo_id, sha, max_index) VALUES (?, ?, 0)", repoID, sha)
 		res, err = e.Exec("UPDATE `commit_status_index` SET max_index=max_index+1 WHERE repo_id=? AND sha=?", repoID, sha)
 		if err != nil {
-			return 0, err
+			return 0, fmt.Errorf("update2 failed: %w", err)
 		}
 		affected, err = res.RowsAffected()
 		if err != nil {
-			return 0, err
+			return 0, fmt.Errorf("RowsAffected failed: %w", err)
 		}
 		// if the update still can not update any records, the record must not exist and there must be some errors (insert error)
 		if affected == 0 {
 			if errIns == nil {
 				return 0, errors.New("impossible error when GetNextCommitStatusIndex, insert and update both succeeded but no record is updated")
 			}
-			return 0, errIns
+			return 0, fmt.Errorf("insert failed: %w", errIns)
 		}
 	}
 
 	// now, the new index is in database (protected by the transaction and write-lock)
 	var newIdx int64
 	has, err := e.SQL("SELECT max_index FROM `commit_status_index` WHERE repo_id=? AND sha=?", repoID, sha).Get(&newIdx)
 	if err != nil {
-		return 0, err
+		return 0, fmt.Errorf("select failed: %w", err)
 	}
 	if !has {
 		return 0, errors.New("impossible error when GetNextCommitStatusIndex, upsert succeeded but no record can be selected")

tests/integration/api_issue_test.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -106,6 +108,49 @@ func TestAPICreateIssue(t *testing.T) {
 	assert.Equal(t, repoBefore.NumClosedIssues, repoAfter.NumClosedIssues)
 }
 
+func TestAPICreateIssueParallel(t *testing.T) {
+	defer tests.PrepareTestEnv(t)()
+	const body, title = "apiTestBody", "apiTestTitle"
+
+	repoBefore := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 1})
+	owner := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: repoBefore.OwnerID})
+
+	session := loginUser(t, owner.Name)
+	token := getTokenForLoggedInUser(t, session, auth_model.AccessTokenScopeRepo)
+	urlStr := fmt.Sprintf("/api/v1/repos/%s/%s/issues?state=all&token=%s", owner.Name, repoBefore.Name, token)
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func(parentT *testing.T, i int) {
+			parentT.Run(fmt.Sprintf("ParallelCreateIssue_%d", i), func(t *testing.T) {
+				newTitle := title + strconv.Itoa(i)
+				newBody := body + strconv.Itoa(i)
+				req := NewRequestWithJSON(t, "POST", urlStr, &api.CreateIssueOption{
+					Body:     newBody,
+					Title:    newTitle,
+					Assignee: owner.Name,
+				})
+				resp := MakeRequest(t, req, http.StatusCreated)
+				var apiIssue api.Issue
+				DecodeJSON(t, resp, &apiIssue)
+				assert.Equal(t, newBody, apiIssue.Body)
+				assert.Equal(t, newTitle, apiIssue.Title)
+
+				unittest.AssertExistsAndLoadBean(t, &issues_model.Issue{
+					RepoID:     repoBefore.ID,
+					AssigneeID: owner.ID,
+					Content:    newBody,
+					Title:      newTitle,
+				})
+
+				wg.Done()
+			})
+		}(t, i)
+	}
+	wg.Wait()
+}
+
 func TestAPIEditIssue(t *testing.T) {
 	defer tests.PrepareTestEnv(t)()