Skip to content

Commit b7db20f

Browse files
lunnylunny
committed
merge
1 parent 888384a commit b7db20f

File tree

3 files changed

+67
-69
lines changed

3 files changed

+67
-69
lines changed

‎models/db/index.go

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,6 @@ var (
2424
ErrGetResourceIndexFailed = errors.New("get resource index failed")
2525
)
2626

27-
const (
28-
// MaxDupIndexAttempts max retry times to create index
29-
MaxDupIndexAttempts = 3
30-
)
31-
3227
// SyncMaxResourceIndex sync the max index with the resource
3328
func SyncMaxResourceIndex(ctx context.Context, tableName string, groupID, maxIndex int64) (err error) {
3429
e := GetEngine(ctx)

‎models/git/commit_status.go

Lines changed: 36 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ package git
77
import (
88
"context"
99
"crypto/sha1"
10+
"errors"
1011
"fmt"
1112
"net/url"
1213
"strings"
@@ -49,79 +50,50 @@ func init() {
4950
db.RegisterModel(new(CommitStatusIndex))
5051
}
5152

52-
// upsertCommitStatusIndex the function will not return until it acquires the lock or receives an error.
53-
func upsertCommitStatusIndex(ctx context.Context, repoID int64, sha string) (err error) {
54-
// An atomic UPSERT operation (INSERT/UPDATE) is the only operation
55-
// that ensures that the key is actually locked.
56-
switch {
57-
case setting.Database.UseSQLite3 || setting.Database.UsePostgreSQL:
58-
_, err = db.Exec(ctx, "INSERT INTO `commit_status_index` (repo_id, sha, max_index) "+
59-
"VALUES (?,?,1) ON CONFLICT (repo_id,sha) DO UPDATE SET max_index = `commit_status_index`.max_index+1",
60-
repoID, sha)
61-
case setting.Database.UseMySQL:
62-
_, err = db.Exec(ctx, "INSERT INTO `commit_status_index` (repo_id, sha, max_index) "+
63-
"VALUES (?,?,1) ON DUPLICATE KEY UPDATE max_index = max_index+1",
64-
repoID, sha)
65-
case setting.Database.UseMSSQL:
66-
// https://weblogs.sqlteam.com/dang/2009/01/31/upsert-race-condition-with-merge/
67-
_, err = db.Exec(ctx, "MERGE `commit_status_index` WITH (HOLDLOCK) as target "+
68-
"USING (SELECT ? AS repo_id, ? AS sha) AS src "+
69-
"ON src.repo_id = target.repo_id AND src.sha = target.sha "+
70-
"WHEN MATCHED THEN UPDATE SET target.max_index = target.max_index+1 "+
71-
"WHEN NOT MATCHED THEN INSERT (repo_id, sha, max_index) "+
72-
"VALUES (src.repo_id, src.sha, 1);",
73-
repoID, sha)
74-
default:
75-
return fmt.Errorf("database type not supported")
76-
}
77-
return err
78-
}
79-
8053
// GetNextCommitStatusIndex retried 3 times to generate a resource index
81-
func GetNextCommitStatusIndex(repoID int64, sha string) (int64, error) {
82-
for i := 0; i < db.MaxDupIndexAttempts; i++ {
83-
idx, err := getNextCommitStatusIndex(repoID, sha)
84-
if err == db.ErrResouceOutdated {
85-
continue
86-
}
87-
if err != nil {
88-
return 0, err
89-
}
90-
return idx, nil
91-
}
92-
return 0, db.ErrGetResourceIndexFailed
93-
}
54+
func GetNextCommitStatusIndex(ctx context.Context, repoID int64, sha string) (int64, error) {
55+
e := db.GetEngine(ctx)
9456

95-
// getNextCommitStatusIndex return the next index
96-
func getNextCommitStatusIndex(repoID int64, sha string) (int64, error) {
97-
ctx, commiter, err := db.TxContext()
57+
// try to update the max_index to next value, and acquire the write-lock for the record
58+
res, err := e.Exec("UPDATE `commit_status_index` SET max_index=max_index+1 WHERE repo_id=? AND sha=?", repoID, sha)
9859
if err != nil {
9960
return 0, err
10061
}
101-
defer commiter.Close()
102-
103-
var preIdx int64
104-
_, err = db.GetEngine(ctx).SQL("SELECT max_index FROM `commit_status_index` WHERE repo_id = ? AND sha = ?", repoID, sha).Get(&preIdx)
62+
affected, err := res.RowsAffected()
10563
if err != nil {
10664
return 0, err
10765
}
66+
if affected == 0 {
67+
// this slow path is only for the first time of creating a resource index
68+
_, errIns := e.Exec("INSERT INTO `commit_status_index` (repo_id, sha, max_index) VALUES (?, ?, 0)", repoID, sha)
69+
res, err = e.Exec("UPDATE `commit_status_index` SET max_index=max_index+1 WHERE repo_id=? AND sha=?", repoID, sha)
70+
if err != nil {
71+
return 0, err
72+
}
10873

109-
if err := upsertCommitStatusIndex(ctx, repoID, sha); err != nil {
110-
return 0, err
74+
affected, err = res.RowsAffected()
75+
if err != nil {
76+
return 0, err
77+
}
78+
// if the update still can not update any records, the record must not exist and there must be some errors (insert error)
79+
if affected == 0 {
80+
if errIns == nil {
81+
return 0, errors.New("impossible error when GetNextCommitStatusIndex, insert and update both succeeded but no record is updated")
82+
}
83+
return 0, errIns
84+
}
11185
}
11286

113-
var curIdx int64
114-
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)
87+
// now, the new index is in database (protected by the transaction and write-lock)
88+
var newIdx int64
89+
has, err := e.SQL("SELECT max_index FROM `commit_status_index` WHERE repo_id=? AND sha=?", repoID, sha).Get(&newIdx)
11590
if err != nil {
11691
return 0, err
11792
}
11893
if !has {
119-
return 0, db.ErrResouceOutdated
120-
}
121-
if err := commiter.Commit(); err != nil {
122-
return 0, err
94+
return 0, errors.New("impossible error when GetNextCommitStatusIndex, upsert succeeded but no record can be selected")
12395
}
124-
return curIdx, nil
96+
return newIdx, nil
12597
}
12698

12799
func (status *CommitStatus) loadAttributes(ctx context.Context) (err error) {
@@ -291,18 +263,18 @@ func NewCommitStatus(opts NewCommitStatusOptions) error {
291263
return fmt.Errorf("NewCommitStatus[%s, %s]: no user specified", repoPath, opts.SHA)
292264
}
293265

294-
// Get the next Status Index
295-
idx, err := GetNextCommitStatusIndex(opts.Repo.ID, opts.SHA)
296-
if err != nil {
297-
return fmt.Errorf("generate commit status index failed: %w", err)
298-
}
299-
300266
ctx, committer, err := db.TxContext()
301267
if err != nil {
302268
return fmt.Errorf("NewCommitStatus[repo_id: %d, user_id: %d, sha: %s]: %w", opts.Repo.ID, opts.Creator.ID, opts.SHA, err)
303269
}
304270
defer committer.Close()
305271

272+
// Get the next Status Index
273+
idx, err := GetNextCommitStatusIndex(ctx, opts.Repo.ID, opts.SHA)
274+
if err != nil {
275+
return fmt.Errorf("generate commit status index failed: %w", err)
276+
}
277+
306278
opts.CommitStatus.Description = strings.TrimSpace(opts.CommitStatus.Description)
307279
opts.CommitStatus.Context = strings.TrimSpace(opts.CommitStatus.Context)
308280
opts.CommitStatus.TargetURL = strings.TrimSpace(opts.CommitStatus.TargetURL)
@@ -316,7 +288,7 @@ func NewCommitStatus(opts NewCommitStatusOptions) error {
316288

317289
// Insert new CommitStatus
318290
if _, err = db.GetEngine(ctx).Insert(opts.CommitStatus); err != nil {
319-
return fmt.Errorf("Insert CommitStatus[%s, %s]: %w", repoPath, opts.SHA, err)
291+
return fmt.Errorf("insert CommitStatus[%s, %s]: %w", repoPath, opts.SHA, err)
320292
}
321293

322294
return committer.Commit()

‎tests/integration/repo_commits_test.go

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,11 @@
55
package integration
66

77
import (
8+
"fmt"
89
"net/http"
910
"net/http/httptest"
1011
"path"
12+
"sync"
1113
"testing"
1214

1315
"code.gitea.io/gitea/modules/json"
@@ -115,3 +117,32 @@ func TestRepoCommitsWithStatusFailure(t *testing.T) {
115117
func TestRepoCommitsWithStatusWarning(t *testing.T) {
116118
doTestRepoCommitWithStatus(t, "warning", "gitea-exclamation", "yellow")
117119
}
120+
121+
func TestRepoCommitsStatusParallel(t *testing.T) {
122+
defer tests.PrepareTestEnv(t)()
123+
124+
session := loginUser(t, "user2")
125+
126+
// Request repository commits page
127+
req := NewRequest(t, "GET", "/user2/repo1/commits/branch/master")
128+
resp := session.MakeRequest(t, req, http.StatusOK)
129+
130+
doc := NewHTMLParser(t, resp.Body)
131+
// Get first commit URL
132+
commitURL, exists := doc.doc.Find("#commits-table tbody tr td.sha a").Attr("href")
133+
assert.True(t, exists)
134+
assert.NotEmpty(t, commitURL)
135+
136+
var wg sync.WaitGroup
137+
for i := 0; i < 10; i++ {
138+
wg.Add(1)
139+
go func(t *testing.T, i int) {
140+
t.Run(fmt.Sprintf("ParallelCreateStatus_%d", i), func(t *testing.T) {
141+
runBody := doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState("pending"))
142+
runBody(t)
143+
wg.Done()
144+
})
145+
}(t, i)
146+
}
147+
wg.Wait()
148+
}

0 commit comments

Comments
 (0)