From 12519e849e3654d9db1ab8ff5a8b91326215c34d Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Wed, 23 Nov 2022 18:18:10 +0800 Subject: [PATCH 1/7] Add test for parallel creating commit status tests --- tests/integration/repo_commits_test.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/integration/repo_commits_test.go b/tests/integration/repo_commits_test.go index c9e77535962b4..fa38f650d3612 100644 --- a/tests/integration/repo_commits_test.go +++ b/tests/integration/repo_commits_test.go @@ -5,9 +5,11 @@ package integration import ( + "fmt" "net/http" "net/http/httptest" "path" + "sync" "testing" "code.gitea.io/gitea/modules/json" @@ -51,6 +53,18 @@ func doTestRepoCommitWithStatus(t *testing.T, state string, classes ...string) { // Call API to add status for commit t.Run("CreateStatus", doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState(state))) + runBody := doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState(state)) + var wg sync.WaitGroup + for i := 0; i < 10; i++ { + wg.Add(1) + t.Run(fmt.Sprintf("ParallelCreateStatus_%d", i), func(t *testing.T) { + t.Parallel() + runBody(t) + wg.Done() + }) + } + wg.Wait() + req = NewRequest(t, "GET", "/user2/repo1/commits/branch/master") resp = session.MakeRequest(t, req, http.StatusOK) From d0408c4ecd8ec3cf5ebbd474cc79c3dcdaf9cb05 Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Fri, 25 Nov 2022 00:08:05 +0800 Subject: [PATCH 2/7] ignore sqlite3 --- tests/integration/repo_commits_test.go | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/tests/integration/repo_commits_test.go b/tests/integration/repo_commits_test.go index fa38f650d3612..898de67ba4d02 100644 --- a/tests/integration/repo_commits_test.go +++ b/tests/integration/repo_commits_test.go @@ -53,17 +53,19 @@ func doTestRepoCommitWithStatus(t *testing.T, state string, classes ...string) { // Call API to add status for commit t.Run("CreateStatus", doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState(state))) - runBody := doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState(state)) - var wg sync.WaitGroup - for i := 0; i < 10; i++ { - wg.Add(1) - t.Run(fmt.Sprintf("ParallelCreateStatus_%d", i), func(t *testing.T) { - t.Parallel() - runBody(t) - wg.Done() - }) + if !setting.Database.UseSQLite3 { + runBody := doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState(state)) + var wg sync.WaitGroup + for i := 0; i < 10; i++ { + wg.Add(1) + t.Run(fmt.Sprintf("ParallelCreateStatus_%d", i), func(t *testing.T) { + t.Parallel() + runBody(t) + wg.Done() + }) + } + wg.Wait() } - wg.Wait() req = NewRequest(t, "GET", "/user2/repo1/commits/branch/master") resp = session.MakeRequest(t, req, http.StatusOK) From a1e34188dd2f2659c54163574eda9a1b21607ed2 Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Fri, 25 Nov 2022 18:41:49 +0800 Subject: [PATCH 3/7] Fix test --- tests/integration/repo_commits_test.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/integration/repo_commits_test.go b/tests/integration/repo_commits_test.go index 898de67ba4d02..fc3883a7feef7 100644 --- a/tests/integration/repo_commits_test.go +++ b/tests/integration/repo_commits_test.go @@ -59,9 +59,10 @@ func doTestRepoCommitWithStatus(t *testing.T, state string, classes ...string) { for i := 0; i < 10; i++ { wg.Add(1) t.Run(fmt.Sprintf("ParallelCreateStatus_%d", i), func(t *testing.T) { - t.Parallel() - runBody(t) - wg.Done() + go func(t *testing.T) { + runBody(t) + wg.Done() + }(t) }) } wg.Wait() From 0fb567718466c47f2946c144f42828a7949eab3b Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Tue, 29 Nov 2022 18:03:19 +0800 Subject: [PATCH 4/7] Fix test --- tests/integration/repo_commits_test.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/integration/repo_commits_test.go b/tests/integration/repo_commits_test.go index fc3883a7feef7..84ce08d62706b 100644 --- a/tests/integration/repo_commits_test.go +++ b/tests/integration/repo_commits_test.go @@ -54,16 +54,16 @@ func doTestRepoCommitWithStatus(t *testing.T, state string, classes ...string) { t.Run("CreateStatus", doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState(state))) if !setting.Database.UseSQLite3 { - runBody := doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState(state)) var wg sync.WaitGroup for i := 0; i < 10; i++ { wg.Add(1) - t.Run(fmt.Sprintf("ParallelCreateStatus_%d", i), func(t *testing.T) { - go func(t *testing.T) { + 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(state)) runBody(t) wg.Done() - }(t) - }) + }) + }(t, i) } wg.Wait() } From 0d4c6b308c9cbf9af08d06a393f2e03f7cc9768b Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Tue, 29 Nov 2022 20:00:08 +0800 Subject: [PATCH 5/7] Fix bug --- models/git/commit_status.go | 73 +++++++++++++------------- tests/integration/repo_commits_test.go | 2 +- 2 files changed, 37 insertions(+), 38 deletions(-) diff --git a/models/git/commit_status.go b/models/git/commit_status.go index 411fbbe536041..bac4aadea0db4 100644 --- a/models/git/commit_status.go +++ b/models/git/commit_status.go @@ -6,6 +6,7 @@ package git import ( "context" "crypto/sha1" + "errors" "fmt" "net/url" "strings" @@ -77,50 +78,48 @@ func upsertCommitStatusIndex(ctx context.Context, repoID int64, sha string) (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 - } - if err := commiter.Commit(); err != nil { - return 0, err + return 0, errors.New("impossible error when GetNextCommitStatusIndex, upsert succeeded but no record can be selected") } - return curIdx, nil + return newIdx, nil } func (status *CommitStatus) loadAttributes(ctx context.Context) (err error) { @@ -290,18 +289,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 +314,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() diff --git a/tests/integration/repo_commits_test.go b/tests/integration/repo_commits_test.go index de580c5432d4d..5b67d5d60e446 100644 --- a/tests/integration/repo_commits_test.go +++ b/tests/integration/repo_commits_test.go @@ -99,7 +99,7 @@ func testRepoCommitsWithStatus(t *testing.T, resp, respOne *httptest.ResponseRec assert.NoError(t, json.Unmarshal(respOne.Body.Bytes(), &status)) assert.NotNil(t, status) - if assert.Len(t, statuses, 1) { + if assert.Len(t, statuses, 11) { assert.Equal(t, api.CommitStatusState(state), statuses[0].State) assert.Equal(t, setting.AppURL+"api/v1/repos/user2/repo1/statuses/65f1bf27bc3bf70f64657658635e66094edbcb4d", statuses[0].URL) assert.Equal(t, "http://test.ci/", statuses[0].TargetURL) From bf50a9118229d47339f56187c04d02b39d8774e8 Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Tue, 29 Nov 2022 20:07:55 +0800 Subject: [PATCH 6/7] remove sqlite condition in test --- models/git/commit_status.go | 28 ---------------- tests/integration/repo_commits_test.go | 46 +++++++++++++++++--------- 2 files changed, 30 insertions(+), 44 deletions(-) diff --git a/models/git/commit_status.go b/models/git/commit_status.go index bac4aadea0db4..6353bb2fa9f97 100644 --- a/models/git/commit_status.go +++ b/models/git/commit_status.go @@ -49,34 +49,6 @@ 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(ctx context.Context, repoID int64, sha string) (int64, error) { e := db.GetEngine(ctx) diff --git a/tests/integration/repo_commits_test.go b/tests/integration/repo_commits_test.go index 5b67d5d60e446..3840c508dcebc 100644 --- a/tests/integration/repo_commits_test.go +++ b/tests/integration/repo_commits_test.go @@ -52,21 +52,6 @@ func doTestRepoCommitWithStatus(t *testing.T, state string, classes ...string) { // Call API to add status for commit t.Run("CreateStatus", doAPICreateCommitStatus(NewAPITestContext(t, "user2", "repo1"), path.Base(commitURL), api.CommitStatusState(state))) - if !setting.Database.UseSQLite3 { - 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(state)) - runBody(t) - wg.Done() - }) - }(t, i) - } - wg.Wait() - } - req = NewRequest(t, "GET", "/user2/repo1/commits/branch/master") resp = session.MakeRequest(t, req, http.StatusOK) @@ -99,7 +84,7 @@ func testRepoCommitsWithStatus(t *testing.T, resp, respOne *httptest.ResponseRec assert.NoError(t, json.Unmarshal(respOne.Body.Bytes(), &status)) assert.NotNil(t, status) - if assert.Len(t, statuses, 11) { + if assert.Len(t, statuses, 1) { assert.Equal(t, api.CommitStatusState(state), statuses[0].State) assert.Equal(t, setting.AppURL+"api/v1/repos/user2/repo1/statuses/65f1bf27bc3bf70f64657658635e66094edbcb4d", statuses[0].URL) assert.Equal(t, "http://test.ci/", statuses[0].TargetURL) @@ -131,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() +} From a46e7e2c7d836c8e36d89f567d9f795194d4d50e Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Tue, 29 Nov 2022 20:43:49 +0800 Subject: [PATCH 7/7] Remove unused const variable --- models/db/index.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/models/db/index.go b/models/db/index.go index 46be74e91e7f6..ca664c9cf125a 100644 --- a/models/db/index.go +++ b/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)