Fix parallel creating commit status bug with tests (#21911) (#21989)

backport #21911 
backport #21998

Co-authored-by: silverwind <me@silverwind.io>
tokarchuk/v1.18
Lunny Xiao 2 years ago committed by GitHub
parent 079ef56824
commit c36a1bc766
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 25
      models/db/index.go
  2. 110
      models/git/commit_status.go
  3. 31
      tests/integration/repo_commits_test.go

@ -8,6 +8,9 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"strconv"
"code.gitea.io/gitea/modules/setting"
) )
// ResourceIndex represents a resource index which could be used as issue/release and others // ResourceIndex represents a resource index which could be used as issue/release and others
@ -24,11 +27,6 @@ var (
ErrGetResourceIndexFailed = errors.New("get resource index failed") 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 // SyncMaxResourceIndex sync the max index with the resource
func SyncMaxResourceIndex(ctx context.Context, tableName string, groupID, maxIndex int64) (err error) { func SyncMaxResourceIndex(ctx context.Context, tableName string, groupID, maxIndex int64) (err error) {
e := GetEngine(ctx) e := GetEngine(ctx)
@ -61,8 +59,25 @@ func SyncMaxResourceIndex(ctx context.Context, tableName string, groupID, maxInd
return nil return nil
} }
func postgresGetNextResourceIndex(ctx context.Context, tableName string, groupID int64) (int64, error) {
res, err := GetEngine(ctx).Query(fmt.Sprintf("INSERT INTO %s (group_id, max_index) "+
"VALUES (?,1) ON CONFLICT (group_id) DO UPDATE SET max_index = %s.max_index+1 RETURNING max_index",
tableName, tableName), groupID)
if err != nil {
return 0, err
}
if len(res) == 0 {
return 0, ErrGetResourceIndexFailed
}
return strconv.ParseInt(string(res[0]["max_index"]), 10, 64)
}
// GetNextResourceIndex generates a resource index, it must run in the same transaction where the resource is created // 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) { func GetNextResourceIndex(ctx context.Context, tableName string, groupID int64) (int64, error) {
if setting.Database.UsePostgreSQL {
return postgresGetNextResourceIndex(ctx, tableName, groupID)
}
e := GetEngine(ctx) e := GetEngine(ctx)
// try to update the max_index to next value, and acquire the write-lock for the record // try to update the max_index to next value, and acquire the write-lock for the record

@ -7,8 +7,10 @@ package git
import ( import (
"context" "context"
"crypto/sha1" "crypto/sha1"
"errors"
"fmt" "fmt"
"net/url" "net/url"
"strconv"
"strings" "strings"
"time" "time"
@ -49,79 +51,67 @@ func init() {
db.RegisterModel(new(CommitStatusIndex)) db.RegisterModel(new(CommitStatusIndex))
} }
// upsertCommitStatusIndex the function will not return until it acquires the lock or receives an error. func postgresGetCommitStatusIndex(ctx context.Context, repoID int64, sha string) (int64, error) {
func upsertCommitStatusIndex(ctx context.Context, repoID int64, sha string) (err error) { res, err := db.GetEngine(ctx).Query("INSERT INTO `commit_status_index` (repo_id, sha, max_index) "+
// An atomic UPSERT operation (INSERT/UPDATE) is the only operation "VALUES (?,?,1) ON CONFLICT (repo_id, sha) DO UPDATE SET max_index = `commit_status_index`.max_index+1 RETURNING max_index",
// that ensures that the key is actually locked. repoID, sha)
switch { if err != nil {
case setting.Database.UseSQLite3 || setting.Database.UsePostgreSQL: return 0, err
_, 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", if len(res) == 0 {
repoID, sha) return 0, db.ErrGetResourceIndexFailed
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 return strconv.ParseInt(string(res[0]["max_index"]), 10, 64)
} }
// GetNextCommitStatusIndex retried 3 times to generate a resource index // GetNextCommitStatusIndex retried 3 times to generate a resource index
func GetNextCommitStatusIndex(repoID int64, sha string) (int64, error) { func GetNextCommitStatusIndex(ctx context.Context, repoID int64, sha string) (int64, error) {
for i := 0; i < db.MaxDupIndexAttempts; i++ { if setting.Database.UsePostgreSQL {
idx, err := getNextCommitStatusIndex(repoID, sha) return postgresGetCommitStatusIndex(ctx, repoID, sha)
if err == db.ErrResouceOutdated {
continue
}
if err != nil {
return 0, err
}
return idx, nil
} }
return 0, db.ErrGetResourceIndexFailed
}
// getNextCommitStatusIndex return the next index e := db.GetEngine(ctx)
func getNextCommitStatusIndex(repoID int64, sha string) (int64, error) {
ctx, commiter, err := db.TxContext() // 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 { if err != nil {
return 0, err return 0, err
} }
defer commiter.Close() affected, err := res.RowsAffected()
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 { if err != nil {
return 0, err 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
}
if err := upsertCommitStatusIndex(ctx, repoID, sha); err != nil { affected, err = res.RowsAffected()
return 0, err 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 // now, the new index is in database (protected by the transaction and write-lock)
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) 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 { if err != nil {
return 0, err return 0, err
} }
if !has { 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) { func (status *CommitStatus) loadAttributes(ctx context.Context) (err error) {
@ -291,18 +281,18 @@ func NewCommitStatus(opts NewCommitStatusOptions) error {
return fmt.Errorf("NewCommitStatus[%s, %s]: no user specified", repoPath, opts.SHA) 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() ctx, committer, err := db.TxContext()
if err != nil { 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) 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() 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.Description = strings.TrimSpace(opts.CommitStatus.Description)
opts.CommitStatus.Context = strings.TrimSpace(opts.CommitStatus.Context) opts.CommitStatus.Context = strings.TrimSpace(opts.CommitStatus.Context)
opts.CommitStatus.TargetURL = strings.TrimSpace(opts.CommitStatus.TargetURL) opts.CommitStatus.TargetURL = strings.TrimSpace(opts.CommitStatus.TargetURL)
@ -316,7 +306,7 @@ func NewCommitStatus(opts NewCommitStatusOptions) error {
// Insert new CommitStatus // Insert new CommitStatus
if _, err = db.GetEngine(ctx).Insert(opts.CommitStatus); err != nil { 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() return committer.Commit()

@ -5,9 +5,11 @@
package integration package integration
import ( import (
"fmt"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"path" "path"
"sync"
"testing" "testing"
"code.gitea.io/gitea/modules/json" "code.gitea.io/gitea/modules/json"
@ -115,3 +117,32 @@ func TestRepoCommitsWithStatusFailure(t *testing.T) {
func TestRepoCommitsWithStatusWarning(t *testing.T) { func TestRepoCommitsWithStatusWarning(t *testing.T) {
doTestRepoCommitWithStatus(t, "warning", "gitea-exclamation", "yellow") 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(parentT *testing.T, i int) {
parentT.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…
Cancel
Save