Sequencial issue index numbering with pessimistic locking mechanism

integrations/locked_resource_test.go

@@ -0,0 +1,125 @@
+// Copyright 2020 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 integrations
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"code.gitea.io/gitea/models"
+	"code.gitea.io/gitea/modules/log"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	// The tests will fail if the waiter function takes less than
+	// blockerDelay minus tolerance to complete.
+	// Note: these values might require tuning in order to avoid
+	// false negatives.
+	waiterDelay  = 100 * time.Millisecond
+	blockerDelay = 200 * time.Millisecond
+	tolerance    = 50 * time.Millisecond // Should be <= (blockerDelay-waiterDelay)/2
+)
+
+type waitResult struct {
+	Waited time.Duration
+	Err    error
+}
+
+func TestLockedResource(t *testing.T) {
+	defer prepareTestEnv(t)()
+
+	// We need to check whether two goroutines block each other
+	// Sadly, there's no way to ensure the second goroutine is
+	// waiting other than using a time delay. The longer the delay,
+	// the more certain we are the second goroutine is waiting.
+
+	// This check **must** fail as we're not blocking anything
+	assert.Error(t, blockTest("no block", func(ctx models.DBContext) error {
+		return nil
+	}))
+
+	models.AssertNotExistsBean(t, &models.LockedResource{LockType: "test-1", LockKey: 1})
+
+	// Test with creation (i.e. new lock type)
+	assert.NoError(t, blockTest("block-new", func(ctx models.DBContext) error {
+		_, err := models.GetLockedResourceCtx(ctx, "block-test-1", 1)
+		return err
+	}))
+
+	// Test without creation (i.e. lock type already exists)
+	assert.NoError(t, blockTest("block-existing", func(ctx models.DBContext) error {
+		_, err := models.GetLockedResourceCtx(ctx, "block-test-1", 1)
+		return err
+	}))
+
+	// Test with temporary record
+	assert.NoError(t, blockTest("block-temp", func(ctx models.DBContext) error {
+		return models.TemporarilyLockResourceKeyCtx(ctx, "temp-1", 1)
+	}))
+}
+
+func blockTest(name string, f func(ctx models.DBContext) error) error {
+	cb := make(chan waitResult)
+	cw := make(chan waitResult)
+	ref := time.Now()
+
+	go func() {
+		cb <- blockTestFunc(name, true, ref, f)
+	}()
+	go func() {
+		cw <- blockTestFunc(name, false, ref, f)
+	}()
+
+	resb := <-cb
+	resw := <-cw
+	if resb.Err != nil {
+		return resb.Err
+	}
+	if resw.Err != nil {
+		return resw.Err
+	}
+
+	if resw.Waited < blockerDelay-tolerance {
+		return fmt.Errorf("Waiter not blocked on %s; wait: %d ms, expected > %d ms",
+			name, resw.Waited.Milliseconds(), (blockerDelay - tolerance).Milliseconds())
+	}
+
+	return nil
+}
+
+func blockTestFunc(name string, blocker bool, ref time.Time, f func(ctx models.DBContext) error) (wr waitResult) {
+	if blocker {
+		name = fmt.Sprintf("blocker [%s]", name)
+	} else {
+		name = fmt.Sprintf("waiter [%s]", name)
+	}
+	err := models.WithTx(func(ctx models.DBContext) error {
+		log.Trace("Entering %s @%d", name, time.Since(ref).Milliseconds())
+		if !blocker {
+			log.Trace("Waiting on %s @%d", name, time.Since(ref).Milliseconds())
+			time.Sleep(waiterDelay)
+			log.Trace("Wait finished on %s @%d", name, time.Since(ref).Milliseconds())
+		}
+		if err := f(ctx); err != nil {
+			return err
+		}
+		if blocker {
+			log.Trace("Waiting on %s @%d", name, time.Since(ref).Milliseconds())
+			time.Sleep(blockerDelay)
+			log.Trace("Wait finished on %s @%d", name, time.Since(ref).Milliseconds())
+		} else {
+			wr.Waited = time.Since(ref)
+		}
+		log.Trace("Finishing %s @%d", name, time.Since(ref).Milliseconds())
+		return nil
+	})
+	if err != nil {
+		wr.Err = fmt.Errorf("error in %s: %v", name, err)
+	}
+	return
+}

models/error.go

@@ -1171,21 +1171,6 @@ func (err ErrIssueLabelTemplateLoad) Error() string {
 	return fmt.Sprintf("Failed to load label template file '%s': %v", err.TemplateFile, err.OriginalError)
 }
 
-// ErrNewIssueInsert is used when the INSERT statement in newIssue fails
-type ErrNewIssueInsert struct {
-	OriginalError error
-}
-
-// IsErrNewIssueInsert checks if an error is a ErrNewIssueInsert.
-func IsErrNewIssueInsert(err error) bool {
-	_, ok := err.(ErrNewIssueInsert)
-	return ok
-}
-
-func (err ErrNewIssueInsert) Error() string {
-	return err.OriginalError.Error()
-}
-
 // ErrIssueWasClosed is used when close a closed issue
 type ErrIssueWasClosed struct {
 	ID    int64

models/issue.go

@@ -75,7 +75,10 @@ var (
 
 const issueTasksRegexpStr = `(^\s*[-*]\s\[[\sx]\]\s.)|(\n\s*[-*]\s\[[\sx]\]\s.)`
 const issueTasksDoneRegexpStr = `(^\s*[-*]\s\[[x]\]\s.)|(\n\s*[-*]\s\[[x]\]\s.)`
-const issueMaxDupIndexAttempts = 3
+
+// IssueLockedEnumerator is the name of the locked_resource used to
+// numerate issues in a repository.
+const IssueLockedEnumerator = "repository-index"
 
 func init() {
 	issueTasksPat = regexp.MustCompile(issueTasksRegexpStr)
@@ -838,19 +841,23 @@ func newIssue(e *xorm.Session, doer *User, opts NewIssueOptions) (err error) {
 	}
 
 	// Milestone validation should happen before insert actual object.
-	if _, err := e.SetExpr("`index`", "coalesce(MAX(`index`),0)+1").
-		Where("repo_id=?", opts.Issue.RepoID).
-		Insert(opts.Issue); err != nil {
-		return ErrNewIssueInsert{err}
-	}
 
-	inserted, err := getIssueByID(e, opts.Issue.ID)
+	// Obtain the next issue number for this repository, which will be locked
+	// and reserved for the remaining of the transaction. Should the transaction
+	// be rolled back, the previous value will be restored.
+	idxresource, err := GetLockedResource(e, IssueLockedEnumerator, opts.Issue.RepoID)
 	if err != nil {
-		return err
+		return fmt.Errorf("GetLockedResource(%s)", IssueLockedEnumerator)
+	}
+	idxresource.Counter++
+	if err := idxresource.UpdateValue(); err != nil {
+		return fmt.Errorf("locked.UpdateValue(%s)", IssueLockedEnumerator)
 	}
+	opts.Issue.Index = idxresource.Counter
 
-	// Patch Index with the value calculated by the database
-	opts.Issue.Index = inserted.Index
+	if _, err = e.Insert(opts.Issue); err != nil {
+		return err
+	}
 
 	if opts.Issue.MilestoneID > 0 {
 		if _, err = e.Exec("UPDATE `milestone` SET num_issues=num_issues+1 WHERE id=?", opts.Issue.MilestoneID); err != nil {
@@ -928,24 +935,6 @@ func newIssue(e *xorm.Session, doer *User, opts NewIssueOptions) (err error) {
 
 // NewIssue creates new issue with labels for repository.
 func NewIssue(repo *Repository, issue *Issue, labelIDs []int64, uuids []string) (err error) {
-	// Retry several times in case INSERT fails due to duplicate key for (repo_id, index); see #7887
-	i := 0
-	for {
-		if err = newIssueAttempt(repo, issue, labelIDs, uuids); err == nil {
-			return nil
-		}
-		if !IsErrNewIssueInsert(err) {
-			return err
-		}
-		if i++; i == issueMaxDupIndexAttempts {
-			break
-		}
-		log.Error("NewIssue: error attempting to insert the new issue; will retry. Original error: %v", err)
-	}
-	return fmt.Errorf("NewIssue: too many errors attempting to insert the new issue. Last error was: %v", err)
-}
-
-func newIssueAttempt(repo *Repository, issue *Issue, labelIDs []int64, uuids []string) (err error) {
 	sess := x.NewSession()
 	defer sess.Close()
 	if err = sess.Begin(); err != nil {
@@ -958,7 +947,7 @@ func newIssueAttempt(repo *Repository, issue *Issue, labelIDs []int64, uuids []s
 		LabelIDs:    labelIDs,
 		Attachments: uuids,
 	}); err != nil {
-		if IsErrUserDoesNotHaveAccessToRepo(err) || IsErrNewIssueInsert(err) {
+		if IsErrUserDoesNotHaveAccessToRepo(err) {
 			return err
 		}
 		return fmt.Errorf("newIssue: %v", err)

models/issue_test.go

@@ -294,7 +294,7 @@ func TestIssue_SearchIssueIDsByKeyword(t *testing.T) {
 	assert.EqualValues(t, []int64{1}, ids)
 }
 
-func testInsertIssue(t *testing.T, title, content string) {
+func testInsertIssue(t *testing.T, title, content string, idx int64) int64 {
 	repo := AssertExistsAndLoadBean(t, &Repository{ID: 1}).(*Repository)
 	user := AssertExistsAndLoadBean(t, &User{ID: 2}).(*User)
 
@@ -313,18 +313,22 @@ func testInsertIssue(t *testing.T, title, content string) {
 	assert.True(t, has)
 	assert.EqualValues(t, issue.Title, newIssue.Title)
 	assert.EqualValues(t, issue.Content, newIssue.Content)
-	// there are 5 issues and max index is 5 on repository 1, so this one should 6
-	assert.EqualValues(t, 6, newIssue.Index)
+	assert.EqualValues(t, idx, newIssue.Index)
 
-	_, err = x.ID(issue.ID).Delete(new(Issue))
-	assert.NoError(t, err)
+	return issue.ID
 }
 
 func TestIssue_InsertIssue(t *testing.T) {
 	assert.NoError(t, PrepareTestDatabase())
 
-	testInsertIssue(t, "my issue1", "special issue's comments?")
-	testInsertIssue(t, `my issue2, this is my son's love \n \r \ `, "special issue's '' comments?")
+	// there are 5 issues and max index is 5 on repository 1, so this one should be 6
+	created := testInsertIssue(t, "my issue1", "special issue's comments?", 6)
+
+	_, err := x.ID(created).Delete(new(Issue))
+	assert.NoError(t, err)
+
+	// deleting an issue should not let a new issue reuse its index number; this one should be 7
+	_ = testInsertIssue(t, `my issue2, this is my son's love \n \r \ `, "special issue's '' comments?", 7)
 }
 
 func TestIssue_ResolveMentions(t *testing.T) {

models/issue_xref_test.go

@@ -130,9 +130,12 @@ func testCreateIssue(t *testing.T, repo, doer int64, title, content string, ispu
 	sess := x.NewSession()
 	defer sess.Close()
 	assert.NoError(t, sess.Begin())
-	_, err := sess.SetExpr("`index`", "coalesce(MAX(`index`),0)+1").Where("repo_id=?", repo).Insert(i)
+	idxresource, err := GetLockedResource(sess, IssueLockedEnumerator, repo)
 	assert.NoError(t, err)
-	i, err = getIssueByID(sess, i.ID)
+	idxresource.Counter++
+	assert.NoError(t, idxresource.UpdateValue())
+	i.Index = idxresource.Counter
+	_, err = sess.Insert(i)
 	assert.NoError(t, err)
 	assert.NoError(t, i.addCrossReferences(sess, d, false))
 	assert.NoError(t, sess.Commit())