Merge branch 'main' into lunny/more_refactor

Author: lunny

models/db/index.go

@@ -23,11 +23,6 @@ var (
 	ErrGetResourceIndexFailed = errors.New("get resource index failed")
 )
 
-const (
-	// MaxDupIndexAttempts max retry times to create index
-	MaxDupIndexAttempts = 3
-)
-
 // SyncMaxResourceIndex sync the max index with the resource
 func SyncMaxResourceIndex(ctx context.Context, tableName string, groupID, maxIndex int64) (err error) {
 	e := GetEngine(ctx)

models/git/commit_status.go

@@ -6,6 +6,7 @@ package git
 import (
 	"context"
 	"crypto/sha1"
+	"errors"
 	"fmt"
 	"net/url"
 	"strings"
@@ -48,79 +49,49 @@ func init() {
 	db.RegisterModel(new(CommitStatusIndex))
 }
 
-// upsertCommitStatusIndex the function will not return until it acquires the lock or receives an error.
-func upsertCommitStatusIndex(ctx context.Context, repoID int64, sha string) (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 = db.Exec(ctx, "INSERT INTO `commit_status_index` (repo_id, sha, max_index) "+
-			"VALUES (?,?,1) ON CONFLICT (repo_id,sha) DO UPDATE SET max_index = `commit_status_index`.max_index+1",
-			repoID, sha)
-	case setting.Database.UseMySQL:
-		_, err = db.Exec(ctx, "INSERT INTO `commit_status_index` (repo_id, sha, max_index) "+
-			"VALUES (?,?,1) ON DUPLICATE KEY UPDATE max_index = max_index+1",
-			repoID, sha)
-	case setting.Database.UseMSSQL:
-		// https://weblogs.sqlteam.com/dang/2009/01/31/upsert-race-condition-with-merge/
-		_, err = db.Exec(ctx, "MERGE `commit_status_index` WITH (HOLDLOCK) as target "+
-			"USING (SELECT ? AS repo_id, ? AS sha) AS src "+
-			"ON src.repo_id = target.repo_id AND src.sha = target.sha "+
-			"WHEN MATCHED THEN UPDATE SET target.max_index = target.max_index+1 "+
-			"WHEN NOT MATCHED THEN INSERT (repo_id, sha, max_index) "+
-			"VALUES (src.repo_id, src.sha, 1);",
-			repoID, sha)
-	default:
-		return fmt.Errorf("database type not supported")
-	}
-	return err
-}
-
 // GetNextCommitStatusIndex retried 3 times to generate a resource index
-func GetNextCommitStatusIndex(repoID int64, sha string) (int64, error) {
-	for i := 0; i < db.MaxDupIndexAttempts; i++ {
-		idx, err := getNextCommitStatusIndex(repoID, sha)
-		if err == db.ErrResouceOutdated {
-			continue
-		}
-		if err != nil {
-			return 0, err
-		}
-		return idx, nil
-	}
-	return 0, db.ErrGetResourceIndexFailed
-}
+func GetNextCommitStatusIndex(ctx context.Context, repoID int64, sha string) (int64, error) {
+	e := db.GetEngine(ctx)
 
-// getNextCommitStatusIndex return the next index
-func getNextCommitStatusIndex(repoID int64, sha string) (int64, error) {
-	ctx, commiter, err := db.TxContext(db.DefaultContext)
+	// 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
 	}
-	defer commiter.Close()
-
-	var preIdx int64
-	_, err = db.GetEngine(ctx).SQL("SELECT max_index FROM `commit_status_index` WHERE repo_id = ? AND sha = ?", repoID, sha).Get(&preIdx)
+	affected, err := res.RowsAffected()
 	if err != nil {
 		return 0, err
 	}
-
-	if err := upsertCommitStatusIndex(ctx, repoID, sha); 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("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
+		}
+		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 GetNextCommitStatusIndex, insert and update both succeeded but no record is updated")
+			}
+			return 0, errIns
+		}
 	}
 
-	var curIdx int64
-	has, err := db.GetEngine(ctx).SQL("SELECT max_index FROM `commit_status_index` WHERE repo_id = ? AND sha = ? AND max_index=?", repoID, sha, 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("SELECT max_index FROM `commit_status_index` WHERE repo_id=? AND sha=?", repoID, sha).Get(&newIdx)
 	if err != nil {
 		return 0, err
 	}
 	if !has {
-		return 0, db.ErrResouceOutdated
+		return 0, errors.New("impossible error when GetNextCommitStatusIndex, upsert succeeded but no record can be selected")
 	}
-	if err := commiter.Commit(); err != nil {
-		return 0, err
-	}
-	return curIdx, nil
+	return newIdx, nil
 }
 
 func (status *CommitStatus) loadAttributes(ctx context.Context) (err error) {
@@ -290,18 +261,18 @@ func NewCommitStatus(opts NewCommitStatusOptions) error {
 		return fmt.Errorf("NewCommitStatus[%s, %s]: no user specified", repoPath, opts.SHA)
 	}
 
-	// Get the next Status Index
-	idx, err := GetNextCommitStatusIndex(opts.Repo.ID, opts.SHA)
-	if err != nil {
-		return fmt.Errorf("generate commit status index failed: %w", err)
-	}
-
 	ctx, committer, err := db.TxContext(db.DefaultContext)
 	if err != nil {
 		return fmt.Errorf("NewCommitStatus[repo_id: %d, user_id: %d, sha: %s]: %w", opts.Repo.ID, opts.Creator.ID, opts.SHA, err)
 	}
 	defer committer.Close()
 
+	// Get the next Status Index
+	idx, err := GetNextCommitStatusIndex(ctx, opts.Repo.ID, opts.SHA)
+	if err != nil {
+		return fmt.Errorf("generate commit status index failed: %w", err)
+	}
+
 	opts.CommitStatus.Description = strings.TrimSpace(opts.CommitStatus.Description)
 	opts.CommitStatus.Context = strings.TrimSpace(opts.CommitStatus.Context)
 	opts.CommitStatus.TargetURL = strings.TrimSpace(opts.CommitStatus.TargetURL)
@@ -315,7 +286,7 @@ func NewCommitStatus(opts NewCommitStatusOptions) error {
 
 	// Insert new CommitStatus
 	if _, err = db.GetEngine(ctx).Insert(opts.CommitStatus); err != nil {
-		return fmt.Errorf("Insert CommitStatus[%s, %s]: %w", repoPath, opts.SHA, err)
+		return fmt.Errorf("insert CommitStatus[%s, %s]: %w", repoPath, opts.SHA, err)
 	}
 
 	return committer.Commit()

modules/storage/helper.go

@@ -4,6 +4,10 @@
 package storage
 
 import (
+	"fmt"
+	"io"
+	"net/url"
+	"os"
 	"reflect"
 
 	"code.gitea.io/gitea/modules/json"
@@ -61,3 +65,31 @@ func toConfig(exemplar, cfg interface{}) (interface{}, error) {
 	}
 	return newVal.Elem().Interface(), nil
 }
+
+var uninitializedStorage = discardStorage("uninitialized storage")
+
+type discardStorage string
+
+func (s discardStorage) Open(_ string) (Object, error) {
+	return nil, fmt.Errorf("%s", s)
+}
+
+func (s discardStorage) Save(_ string, _ io.Reader, _ int64) (int64, error) {
+	return 0, fmt.Errorf("%s", s)
+}
+
+func (s discardStorage) Stat(_ string) (os.FileInfo, error) {
+	return nil, fmt.Errorf("%s", s)
+}
+
+func (s discardStorage) Delete(_ string) error {
+	return fmt.Errorf("%s", s)
+}
+
+func (s discardStorage) URL(_, _ string) (*url.URL, error) {
+	return nil, fmt.Errorf("%s", s)
+}
+
+func (s discardStorage) IterateObjects(_ func(string, Object) error) error {
+	return fmt.Errorf("%s", s)
+}

modules/storage/helper_test.go

@@ -0,0 +1,50 @@
+// Copyright 2022 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package storage
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_discardStorage(t *testing.T) {
+	tests := []discardStorage{
+		uninitializedStorage,
+		discardStorage("empty"),
+	}
+	for _, tt := range tests {
+		t.Run(string(tt), func(t *testing.T) {
+			{
+				got, err := tt.Open("path")
+				assert.Nil(t, got)
+				assert.Error(t, err, string(tt))
+			}
+			{
+				got, err := tt.Save("path", bytes.NewReader([]byte{0}), 1)
+				assert.Equal(t, int64(0), got)
+				assert.Error(t, err, string(tt))
+			}
+			{
+				got, err := tt.Stat("path")
+				assert.Nil(t, got)
+				assert.Error(t, err, string(tt))
+			}
+			{
+				err := tt.Delete("path")
+				assert.Error(t, err, string(tt))
+			}
+			{
+				got, err := tt.URL("path", "name")
+				assert.Nil(t, got)
+				assert.Errorf(t, err, string(tt))
+			}
+			{
+				err := tt.IterateObjects(func(_ string, _ Object) error { return nil })
+				assert.Error(t, err, string(tt))
+			}
+		})
+	}
+}

modules/storage/storage.go

@@ -110,46 +110,38 @@ func SaveFrom(objStorage ObjectStorage, p string, callback func(w io.Writer) err
 
 var (
 	// Attachments represents attachments storage
-	Attachments ObjectStorage
+	Attachments ObjectStorage = uninitializedStorage
 
 	// LFS represents lfs storage
-	LFS ObjectStorage
+	LFS ObjectStorage = uninitializedStorage
 
 	// Avatars represents user avatars storage
-	Avatars ObjectStorage
+	Avatars ObjectStorage = uninitializedStorage
 	// RepoAvatars represents repository avatars storage
-	RepoAvatars ObjectStorage
+	RepoAvatars ObjectStorage = uninitializedStorage
 
 	// RepoArchives represents repository archives storage
-	RepoArchives ObjectStorage
+	RepoArchives ObjectStorage = uninitializedStorage
 
 	// Packages represents packages storage
-	Packages ObjectStorage
+	Packages ObjectStorage = uninitializedStorage
 )
 
 // Init init the stoarge
 func Init() error {
-	if err := initAttachments(); err != nil {
-		return err
-	}
-
-	if err := initAvatars(); err != nil {
-		return err
-	}
-
-	if err := initRepoAvatars(); err != nil {
-		return err
-	}
-
-	if err := initLFS(); err != nil {
-		return err
-	}
-
-	if err := initRepoArchives(); err != nil {
-		return err
+	for _, f := range []func() error{
+		initAttachments,
+		initAvatars,
+		initRepoAvatars,
+		initLFS,
+		initRepoArchives,
+		initPackages,
+	} {
+		if err := f(); err != nil {
+			return err
+		}
 	}
-
-	return initPackages()
+	return nil
 }
 
 // NewStorage takes a storage type and some config and returns an ObjectStorage or an error
@@ -172,12 +164,20 @@ func initAvatars() (err error) {
 }
 
 func initAttachments() (err error) {
+	if !setting.Attachment.Enabled {
+		Attachments = discardStorage("Attachment isn't enabled")
+		return nil
+	}
 	log.Info("Initialising Attachment storage with type: %s", setting.Attachment.Storage.Type)
 	Attachments, err = NewStorage(setting.Attachment.Storage.Type, &setting.Attachment.Storage)
 	return err
 }
 
 func initLFS() (err error) {
+	if !setting.LFS.StartServer {
+		LFS = discardStorage("LFS isn't enabled")
+		return nil
+	}
 	log.Info("Initialising LFS storage with type: %s", setting.LFS.Storage.Type)
 	LFS, err = NewStorage(setting.LFS.Storage.Type, &setting.LFS.Storage)
 	return err
@@ -196,6 +196,10 @@ func initRepoArchives() (err error) {
 }
 
 func initPackages() (err error) {
+	if !setting.Packages.Enabled {
+		Packages = discardStorage("Packages isn't enabled")
+		return nil
+	}
 	log.Info("Initialising Packages storage with type: %s", setting.Packages.Storage.Type)
 	Packages, err = NewStorage(setting.Packages.Storage.Type, &setting.Packages.Storage)
 	return err

tests/integration/repo_commits_test.go

@@ -4,9 +4,11 @@
 package integration
 
 import (
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"path"
+	"sync"
 	"testing"
 
 	"code.gitea.io/gitea/modules/json"
@@ -114,3 +116,32 @@ func TestRepoCommitsWithStatusFailure(t *testing.T) {
 func TestRepoCommitsWithStatusWarning(t *testing.T) {
 	doTestRepoCommitWithStatus(t, "warning", "gitea-exclamation", "yellow")
 }
+
+func TestRepoCommitsStatusParallel(t *testing.T) {
+	defer tests.PrepareTestEnv(t)()
+
+	session := loginUser(t, "user2")
+
+	// Request repository commits page
+	req := NewRequest(t, "GET", "/user2/repo1/commits/branch/master")
+	resp := session.MakeRequest(t, req, http.StatusOK)
+
+	doc := NewHTMLParser(t, resp.Body)
+	// Get first commit URL
+	commitURL, exists := doc.doc.Find("#commits-table tbody tr td.sha a").Attr("href")
+	assert.True(t, exists)
+	assert.NotEmpty(t, commitURL)
+
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func(t *testing.T, i int) {
+			t.Run(fmt.Sprintf("ParallelCreateStatus_%d", i), func(t *testing.T) {
+				runBody := doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState("pending"))
+				runBody(t)
+				wg.Done()
+			})
+		}(t, i)
+	}
+	wg.Wait()
+}