Refactor GetNextResourceIndex to make it work properly with transaction (#21469)

Related:
* #21362

This PR uses a general and stable method to generate resource index (eg:
Issue Index, PR Index)

If the code looks good, I can add more tests

ps: please skip the diff, only have a look at the new code. It's
entirely re-written.

Co-authored-by: Lunny Xiao <[email protected]>

Author: wxiaoguang

models/db/index.go

@@ -8,45 +8,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
-
-	"code.gitea.io/gitea/modules/setting"
 )
 
 // ResourceIndex represents a resource index which could be used as issue/release and others
-// We can create different tables i.e. issue_index, release_index and etc.
+// We can create different tables i.e. issue_index, release_index, etc.
 type ResourceIndex struct {
 	GroupID  int64 `xorm:"pk"`
 	MaxIndex int64 `xorm:"index"`
 }
 
-// UpsertResourceIndex the function will not return until it acquires the lock or receives an error.
-func UpsertResourceIndex(ctx context.Context, tableName string, groupID int64) (err error) {
-	// An atomic UPSERT operation (INSERT/UPDATE) is the only operation
-	// that ensures that the key is actually locked.
-	switch {
-	case setting.Database.UseSQLite3 || setting.Database.UsePostgreSQL:
-		_, err = Exec(ctx, fmt.Sprintf("INSERT INTO %s (group_id, max_index) "+
-			"VALUES (?,1) ON CONFLICT (group_id) DO UPDATE SET max_index = %s.max_index+1",
-			tableName, tableName), groupID)
-	case setting.Database.UseMySQL:
-		_, err = Exec(ctx, fmt.Sprintf("INSERT INTO %s (group_id, max_index) "+
-			"VALUES (?,1) ON DUPLICATE KEY UPDATE max_index = max_index+1", tableName),
-			groupID)
-	case setting.Database.UseMSSQL:
-		// https://weblogs.sqlteam.com/dang/2009/01/31/upsert-race-condition-with-merge/
-		_, err = Exec(ctx, fmt.Sprintf("MERGE %s WITH (HOLDLOCK) as target "+
-			"USING (SELECT ? AS group_id) AS src "+
-			"ON src.group_id = target.group_id "+
-			"WHEN MATCHED THEN UPDATE SET target.max_index = target.max_index+1 "+
-			"WHEN NOT MATCHED THEN INSERT (group_id, max_index) "+
-			"VALUES (src.group_id, 1);", tableName),
-			groupID)
-	default:
-		return fmt.Errorf("database type not supported")
-	}
-	return err
-}
-
 var (
 	// ErrResouceOutdated represents an error when request resource outdated
 	ErrResouceOutdated = errors.New("resource outdated")
@@ -59,53 +29,85 @@ const (
 	MaxDupIndexAttempts = 3
 )
 
-// GetNextResourceIndex retried 3 times to generate a resource index
-func GetNextResourceIndex(tableName string, groupID int64) (int64, error) {
-	for i := 0; i < MaxDupIndexAttempts; i++ {
-		idx, err := getNextResourceIndex(tableName, groupID)
-		if err == ErrResouceOutdated {
-			continue
-		}
+// SyncMaxResourceIndex sync the max index with the resource
+func SyncMaxResourceIndex(ctx context.Context, tableName string, groupID, maxIndex int64) (err error) {
+	e := GetEngine(ctx)
+
+	// try to update the max_index and acquire the write-lock for the record
+	res, err := e.Exec(fmt.Sprintf("UPDATE %s SET max_index=? WHERE group_id=? AND max_index<?", tableName), maxIndex, groupID, maxIndex)
+	if err != nil {
+		return err
+	}
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return err
+	}
+	if affected == 0 {
+		// if nothing is updated, the record might not exist or might be larger, it's safe to try to insert it again and then check whether the record exists
+		_, errIns := e.Exec(fmt.Sprintf("INSERT INTO %s (group_id, max_index) VALUES (?, ?)", tableName), groupID, maxIndex)
+		var savedIdx int64
+		has, err := e.SQL(fmt.Sprintf("SELECT max_index FROM %s WHERE group_id=?", tableName), groupID).Get(&savedIdx)
 		if err != nil {
-			return 0, err
+			return err
+		}
+		// if the record still doesn't exist, there must be some errors (insert error)
+		if !has {
+			if errIns == nil {
+				return errors.New("impossible error when SyncMaxResourceIndex, insert succeeded but no record is saved")
+			}
+			return errIns
 		}
-		return idx, nil
 	}
-	return 0, ErrGetResourceIndexFailed
+	return nil
 }
 
-// DeleteResouceIndex delete resource index
-func DeleteResouceIndex(ctx context.Context, tableName string, groupID int64) error {
-	_, err := Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE group_id=?", tableName), groupID)
-	return err
-}
+// 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) {
+	e := GetEngine(ctx)
 
-// getNextResourceIndex return the next index
-func getNextResourceIndex(tableName string, groupID int64) (int64, error) {
-	ctx, commiter, err := TxContext()
+	// try to update the max_index to next value, and acquire the write-lock for the record
+	res, err := e.Exec(fmt.Sprintf("UPDATE %s SET max_index=max_index+1 WHERE group_id=?", tableName), groupID)
 	if err != nil {
 		return 0, err
 	}
-	defer commiter.Close()
-	var preIdx int64
-	if _, err := GetEngine(ctx).SQL(fmt.Sprintf("SELECT max_index FROM %s WHERE group_id = ?", tableName), groupID).Get(&preIdx); err != nil {
+	affected, err := res.RowsAffected()
+	if err != nil {
 		return 0, err
 	}
-
-	if err := UpsertResourceIndex(ctx, tableName, groupID); err != nil {
-		return 0, err
+	if affected == 0 {
+		// this slow path is only for the first time of creating a resource index
+		_, errIns := e.Exec(fmt.Sprintf("INSERT INTO %s (group_id, max_index) VALUES (?, 0)", tableName), groupID)
+		res, err = e.Exec(fmt.Sprintf("UPDATE %s SET max_index=max_index+1 WHERE group_id=?", tableName), groupID)
+		if err != nil {
+			return 0, err
+		}
+		affected, err = res.RowsAffected()
+		if err != nil {
+			return 0, 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 GetNextResourceIndex, insert and update both succeeded but no record is updated")
+			}
+			return 0, errIns
+		}
 	}
 
-	var curIdx int64
-	has, err := GetEngine(ctx).SQL(fmt.Sprintf("SELECT max_index FROM %s WHERE group_id = ? AND max_index=?", tableName), groupID, preIdx+1).Get(&curIdx)
+	// now, the new index is in database (protected by the transaction and write-lock)
+	var newIdx int64
+	has, err := e.SQL(fmt.Sprintf("SELECT max_index FROM %s WHERE group_id=?", tableName), groupID).Get(&newIdx)
 	if err != nil {
 		return 0, err
 	}
 	if !has {
-		return 0, ErrResouceOutdated
-	}
-	if err := commiter.Commit(); err != nil {
-		return 0, err
+		return 0, errors.New("impossible error when GetNextResourceIndex, upsert succeeded but no record can be selected")
 	}
-	return curIdx, nil
+	return newIdx, nil
+}
+
+// DeleteResourceIndex delete resource index
+func DeleteResourceIndex(ctx context.Context, tableName string, groupID int64) error {
+	_, err := Exec(ctx, fmt.Sprintf("DELETE FROM %s WHERE group_id=?", tableName), groupID)
+	return err
 }

models/db/index_test.go

@@ -0,0 +1,127 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package db_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"code.gitea.io/gitea/models/db"
+	"code.gitea.io/gitea/models/unittest"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type TestIndex db.ResourceIndex
+
+func getCurrentResourceIndex(ctx context.Context, tableName string, groupID int64) (int64, error) {
+	e := db.GetEngine(ctx)
+	var idx int64
+	has, err := e.SQL(fmt.Sprintf("SELECT max_index FROM %s WHERE group_id=?", tableName), groupID).Get(&idx)
+	if err != nil {
+		return 0, err
+	}
+	if !has {
+		return 0, errors.New("no record")
+	}
+	return idx, nil
+}
+
+func TestSyncMaxResourceIndex(t *testing.T) {
+	assert.NoError(t, unittest.PrepareTestDatabase())
+	xe := unittest.GetXORMEngine()
+	assert.NoError(t, xe.Sync(&TestIndex{}))
+
+	err := db.SyncMaxResourceIndex(db.DefaultContext, "test_index", 10, 51)
+	assert.NoError(t, err)
+
+	// sync new max index
+	maxIndex, err := getCurrentResourceIndex(db.DefaultContext, "test_index", 10)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 51, maxIndex)
+
+	// smaller index doesn't change
+	err = db.SyncMaxResourceIndex(db.DefaultContext, "test_index", 10, 30)
+	assert.NoError(t, err)
+	maxIndex, err = getCurrentResourceIndex(db.DefaultContext, "test_index", 10)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 51, maxIndex)
+
+	// larger index changes
+	err = db.SyncMaxResourceIndex(db.DefaultContext, "test_index", 10, 62)
+	assert.NoError(t, err)
+	maxIndex, err = getCurrentResourceIndex(db.DefaultContext, "test_index", 10)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 62, maxIndex)
+
+	// commit transaction
+	err = db.WithTx(func(ctx context.Context) error {
+		err = db.SyncMaxResourceIndex(ctx, "test_index", 10, 73)
+		assert.NoError(t, err)
+		maxIndex, err = getCurrentResourceIndex(ctx, "test_index", 10)
+		assert.NoError(t, err)
+		assert.EqualValues(t, 73, maxIndex)
+		return nil
+	})
+	assert.NoError(t, err)
+	maxIndex, err = getCurrentResourceIndex(db.DefaultContext, "test_index", 10)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 73, maxIndex)
+
+	// rollback transaction
+	err = db.WithTx(func(ctx context.Context) error {
+		err = db.SyncMaxResourceIndex(ctx, "test_index", 10, 84)
+		maxIndex, err = getCurrentResourceIndex(ctx, "test_index", 10)
+		assert.NoError(t, err)
+		assert.EqualValues(t, 84, maxIndex)
+		return errors.New("test rollback")
+	})
+	assert.Error(t, err)
+	maxIndex, err = getCurrentResourceIndex(db.DefaultContext, "test_index", 10)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 73, maxIndex) // the max index doesn't change because the transaction was rolled back
+}
+
+func TestGetNextResourceIndex(t *testing.T) {
+	assert.NoError(t, unittest.PrepareTestDatabase())
+	xe := unittest.GetXORMEngine()
+	assert.NoError(t, xe.Sync(&TestIndex{}))
+
+	// create a new record
+	maxIndex, err := db.GetNextResourceIndex(db.DefaultContext, "test_index", 20)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 1, maxIndex)
+
+	// increase the existing record
+	maxIndex, err = db.GetNextResourceIndex(db.DefaultContext, "test_index", 20)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 2, maxIndex)
+
+	// commit transaction
+	err = db.WithTx(func(ctx context.Context) error {
+		maxIndex, err = db.GetNextResourceIndex(ctx, "test_index", 20)
+		assert.NoError(t, err)
+		assert.EqualValues(t, 3, maxIndex)
+		return nil
+	})
+	assert.NoError(t, err)
+	maxIndex, err = getCurrentResourceIndex(db.DefaultContext, "test_index", 20)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 3, maxIndex)
+
+	// rollback transaction
+	err = db.WithTx(func(ctx context.Context) error {
+		maxIndex, err = db.GetNextResourceIndex(ctx, "test_index", 20)
+		assert.NoError(t, err)
+		assert.EqualValues(t, 4, maxIndex)
+		return errors.New("test rollback")
+	})
+	assert.Error(t, err)
+	maxIndex, err = getCurrentResourceIndex(db.DefaultContext, "test_index", 20)
+	assert.NoError(t, err)
+	assert.EqualValues(t, 3, maxIndex) // the max index doesn't change because the transaction was rolled back
+}

models/issues/issue.go

@@ -1064,19 +1064,19 @@ func NewIssueWithIndex(ctx context.Context, doer *user_model.User, opts NewIssue
 
 // NewIssue creates new issue with labels for repository.
 func NewIssue(repo *repo_model.Repository, issue *Issue, labelIDs []int64, uuids []string) (err error) {
-	idx, err := db.GetNextResourceIndex("issue_index", repo.ID)
-	if err != nil {
-		return fmt.Errorf("generate issue index failed: %v", err)
-	}
-
-	issue.Index = idx
-
 	ctx, committer, err := db.TxContext()
 	if err != nil {
 		return err
 	}
 	defer committer.Close()
 
+	idx, err := db.GetNextResourceIndex(ctx, "issue_index", repo.ID)
+	if err != nil {
+		return fmt.Errorf("generate issue index failed: %w", err)
+	}
+
+	issue.Index = idx
+
 	if err = NewIssueWithIndex(ctx, issue.Poster, NewIssueOptions{
 		Repo:        repo,
 		Issue:       issue,

models/issues/issue_index.go

@@ -15,16 +15,12 @@ func RecalculateIssueIndexForRepo(repoID int64) error {
 	}
 	defer committer.Close()
 
-	if err := db.UpsertResourceIndex(ctx, "issue_index", repoID); err != nil {
-		return err
-	}
-
 	var max int64
-	if _, err := db.GetEngine(ctx).Select(" MAX(`index`)").Table("issue").Where("repo_id=?", repoID).Get(&max); err != nil {
+	if _, err = db.GetEngine(ctx).Select(" MAX(`index`)").Table("issue").Where("repo_id=?", repoID).Get(&max); err != nil {
 		return err
 	}
 
-	if _, err := db.GetEngine(ctx).Exec("UPDATE `issue_index` SET max_index=? WHERE group_id=?", max, repoID); err != nil {
+	if err = db.SyncMaxResourceIndex(ctx, "issue_index", repoID, max); err != nil {
 		return err
 	}
 

models/issues/issue_xref_test.go

@@ -131,7 +131,11 @@ func testCreateIssue(t *testing.T, repo, doer int64, title, content string, ispu
 	r := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: repo})
 	d := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: doer})
 
-	idx, err := db.GetNextResourceIndex("issue_index", r.ID)
+	ctx, committer, err := db.TxContext()
+	assert.NoError(t, err)
+	defer committer.Close()
+
+	idx, err := db.GetNextResourceIndex(ctx, "issue_index", r.ID)
 	assert.NoError(t, err)
 	i := &issues_model.Issue{
 		RepoID:   r.ID,
@@ -143,9 +147,6 @@ func testCreateIssue(t *testing.T, repo, doer int64, title, content string, ispu
 		Index:    idx,
 	}
 
-	ctx, committer, err := db.TxContext()
-	assert.NoError(t, err)
-	defer committer.Close()
 	err = issues_model.NewIssueWithIndex(ctx, d, issues_model.NewIssueOptions{
 		Repo:  r,
 		Issue: i,

models/issues/pull.go

@@ -490,20 +490,20 @@ func (pr *PullRequest) SetMerged(ctx context.Context) (bool, error) {
 
 // NewPullRequest creates new pull request with labels for repository.
 func NewPullRequest(outerCtx context.Context, repo *repo_model.Repository, issue *Issue, labelIDs []int64, uuids []string, pr *PullRequest) (err error) {
-	idx, err := db.GetNextResourceIndex("issue_index", repo.ID)
-	if err != nil {
-		return fmt.Errorf("generate pull request index failed: %v", err)
-	}
-
-	issue.Index = idx
-
 	ctx, committer, err := db.TxContext()
 	if err != nil {
 		return err
 	}
 	defer committer.Close()
 	ctx.WithContext(outerCtx)
 
+	idx, err := db.GetNextResourceIndex(ctx, "issue_index", repo.ID)
+	if err != nil {
+		return fmt.Errorf("generate pull request index failed: %v", err)
+	}
+
+	issue.Index = idx
+
 	if err = NewIssueWithIndex(ctx, issue.Poster, NewIssueOptions{
 		Repo:        repo,
 		Issue:       issue,

models/repo.go

@@ -163,7 +163,7 @@ func DeleteRepository(doer *user_model.User, uid, repoID int64) error {
 	}
 
 	// Delete issue index
-	if err := db.DeleteResouceIndex(ctx, "issue_index", repoID); err != nil {
+	if err := db.DeleteResourceIndex(ctx, "issue_index", repoID); err != nil {
 		return err
 	}