mirror of
https://github.com/go-gitea/gitea.git
synced 2025-07-20 04:15:18 +02:00
Merge branch 'main' into feature/bots
This commit is contained in:
commit
c5c9b4ae57
@ -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)
|
||||
|
@ -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
|
||||
}
|
||||
func GetNextCommitStatusIndex(ctx context.Context, repoID int64, sha string) (int64, error) {
|
||||
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
|
||||
}
|
||||
affected, err := res.RowsAffected()
|
||||
if 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
|
||||
}
|
||||
return idx, nil
|
||||
}
|
||||
return 0, db.ErrGetResourceIndexFailed
|
||||
}
|
||||
|
||||
// getNextCommitStatusIndex return the next index
|
||||
func getNextCommitStatusIndex(repoID int64, sha string) (int64, error) {
|
||||
ctx, commiter, err := db.TxContext(db.DefaultContext)
|
||||
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)
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
if err := upsertCommitStatusIndex(ctx, repoID, sha); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
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()
|
||||
|
@ -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)
|
||||
}
|
||||
|
50
modules/storage/helper_test.go
Normal file
50
modules/storage/helper_test.go
Normal file
@ -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))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
@ -110,24 +110,24 @@ 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
|
||||
|
||||
// Bots represents bots storage
|
||||
Bots ObjectStorage
|
||||
Bots ObjectStorage = uninitializedStorage
|
||||
)
|
||||
|
||||
// Init init the stoarge
|
||||
@ -168,6 +168,10 @@ 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
|
||||
@ -192,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
|
||||
|
@ -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()
|
||||
}
|
||||
|
Loading…
x
Reference in New Issue
Block a user