1
1
mirror of https://github.com/go-gitea/gitea synced 2024-12-22 16:44:26 +00:00

Refactor push mirror find and add check for updating push mirror (#32539)

Co-authored-by: wxiaoguang <wxiaoguang@gmail.com>
This commit is contained in:
Lunny Xiao 2024-11-17 21:59:04 -08:00 committed by GitHub
parent 8a20fba8eb
commit 696fbe6036
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 141 additions and 107 deletions

View File

@ -68,7 +68,8 @@ func CheckCollations(x *xorm.Engine) (*CheckCollationsResult, error) {
var candidateCollations []string var candidateCollations []string
if x.Dialect().URI().DBType == schemas.MYSQL { if x.Dialect().URI().DBType == schemas.MYSQL {
if _, err = x.SQL("SELECT @@collation_database").Get(&res.DatabaseCollation); err != nil { _, err = x.SQL("SELECT DEFAULT_COLLATION_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME = ?", setting.Database.Name).Get(&res.DatabaseCollation)
if err != nil {
return nil, err return nil, err
} }
res.IsCollationCaseSensitive = func(s string) bool { res.IsCollationCaseSensitive = func(s string) bool {

View File

@ -9,15 +9,13 @@ import (
"code.gitea.io/gitea/models/db" "code.gitea.io/gitea/models/db"
"code.gitea.io/gitea/modules/log" "code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/optional"
"code.gitea.io/gitea/modules/timeutil" "code.gitea.io/gitea/modules/timeutil"
"code.gitea.io/gitea/modules/util" "code.gitea.io/gitea/modules/util"
"xorm.io/builder" "xorm.io/builder"
) )
// ErrPushMirrorNotExist mirror does not exist error
var ErrPushMirrorNotExist = util.NewNotExistErrorf("PushMirror does not exist")
// PushMirror represents mirror information of a repository. // PushMirror represents mirror information of a repository.
type PushMirror struct { type PushMirror struct {
ID int64 `xorm:"pk autoincr"` ID int64 `xorm:"pk autoincr"`
@ -96,26 +94,46 @@ func DeletePushMirrors(ctx context.Context, opts PushMirrorOptions) error {
return util.NewInvalidArgumentErrorf("repoID required and must be set") return util.NewInvalidArgumentErrorf("repoID required and must be set")
} }
type findPushMirrorOptions struct {
db.ListOptions
RepoID int64
SyncOnCommit optional.Option[bool]
}
func (opts findPushMirrorOptions) ToConds() builder.Cond {
cond := builder.NewCond()
if opts.RepoID > 0 {
cond = cond.And(builder.Eq{"repo_id": opts.RepoID})
}
if opts.SyncOnCommit.Has() {
cond = cond.And(builder.Eq{"sync_on_commit": opts.SyncOnCommit.Value()})
}
return cond
}
// GetPushMirrorsByRepoID returns push-mirror information of a repository. // GetPushMirrorsByRepoID returns push-mirror information of a repository.
func GetPushMirrorsByRepoID(ctx context.Context, repoID int64, listOptions db.ListOptions) ([]*PushMirror, int64, error) { func GetPushMirrorsByRepoID(ctx context.Context, repoID int64, listOptions db.ListOptions) ([]*PushMirror, int64, error) {
sess := db.GetEngine(ctx).Where("repo_id = ?", repoID) return db.FindAndCount[PushMirror](ctx, findPushMirrorOptions{
if listOptions.Page != 0 { ListOptions: listOptions,
sess = db.SetSessionPagination(sess, &listOptions) RepoID: repoID,
mirrors := make([]*PushMirror, 0, listOptions.PageSize) })
count, err := sess.FindAndCount(&mirrors) }
return mirrors, count, err
func GetPushMirrorByIDAndRepoID(ctx context.Context, id, repoID int64) (*PushMirror, bool, error) {
var pushMirror PushMirror
has, err := db.GetEngine(ctx).Where("id = ?", id).And("repo_id = ?", repoID).Get(&pushMirror)
if !has || err != nil {
return nil, has, err
} }
mirrors := make([]*PushMirror, 0, 10) return &pushMirror, true, nil
count, err := sess.FindAndCount(&mirrors)
return mirrors, count, err
} }
// GetPushMirrorsSyncedOnCommit returns push-mirrors for this repo that should be updated by new commits // GetPushMirrorsSyncedOnCommit returns push-mirrors for this repo that should be updated by new commits
func GetPushMirrorsSyncedOnCommit(ctx context.Context, repoID int64) ([]*PushMirror, error) { func GetPushMirrorsSyncedOnCommit(ctx context.Context, repoID int64) ([]*PushMirror, error) {
mirrors := make([]*PushMirror, 0, 10) return db.Find[PushMirror](ctx, findPushMirrorOptions{
return mirrors, db.GetEngine(ctx). RepoID: repoID,
Where("repo_id = ? AND sync_on_commit = ?", repoID, true). SyncOnCommit: optional.Some(true),
Find(&mirrors) })
} }
// PushMirrorsIterate iterates all push-mirror repositories. // PushMirrorsIterate iterates all push-mirror repositories.

View File

@ -8,7 +8,6 @@ import (
"errors" "errors"
"fmt" "fmt"
"net/http" "net/http"
"strconv"
"strings" "strings"
"time" "time"
@ -290,8 +289,8 @@ func SettingsPost(ctx *context.Context) {
return return
} }
m, err := selectPushMirrorByForm(ctx, form, repo) m, _, _ := repo_model.GetPushMirrorByIDAndRepoID(ctx, form.PushMirrorID, repo.ID)
if err != nil { if m == nil {
ctx.NotFound("", nil) ctx.NotFound("", nil)
return return
} }
@ -317,15 +316,13 @@ func SettingsPost(ctx *context.Context) {
return return
} }
id, err := strconv.ParseInt(form.PushMirrorID, 10, 64) m, _, _ := repo_model.GetPushMirrorByIDAndRepoID(ctx, form.PushMirrorID, repo.ID)
if err != nil { if m == nil {
ctx.ServerError("UpdatePushMirrorIntervalPushMirrorID", err) ctx.NotFound("", nil)
return return
} }
m := &repo_model.PushMirror{
ID: id, m.Interval = interval
Interval: interval,
}
if err := repo_model.UpdatePushMirrorInterval(ctx, m); err != nil { if err := repo_model.UpdatePushMirrorInterval(ctx, m); err != nil {
ctx.ServerError("UpdatePushMirrorInterval", err) ctx.ServerError("UpdatePushMirrorInterval", err)
return return
@ -334,7 +331,10 @@ func SettingsPost(ctx *context.Context) {
// If we observed its implementation in the context of `push-mirror-sync` where it // If we observed its implementation in the context of `push-mirror-sync` where it
// is evident that pushing to the queue is necessary for updates. // is evident that pushing to the queue is necessary for updates.
// So, there are updates within the given interval, it is necessary to update the queue accordingly. // So, there are updates within the given interval, it is necessary to update the queue accordingly.
if !ctx.FormBool("push_mirror_defer_sync") {
// push_mirror_defer_sync is mainly for testing purpose, we do not really want to sync the push mirror immediately
mirror_service.AddPushMirrorToQueue(m.ID) mirror_service.AddPushMirrorToQueue(m.ID)
}
ctx.Flash.Success(ctx.Tr("repo.settings.update_settings_success")) ctx.Flash.Success(ctx.Tr("repo.settings.update_settings_success"))
ctx.Redirect(repo.Link() + "/settings") ctx.Redirect(repo.Link() + "/settings")
@ -348,18 +348,18 @@ func SettingsPost(ctx *context.Context) {
// as an error on the UI for this action // as an error on the UI for this action
ctx.Data["Err_RepoName"] = nil ctx.Data["Err_RepoName"] = nil
m, err := selectPushMirrorByForm(ctx, form, repo) m, _, _ := repo_model.GetPushMirrorByIDAndRepoID(ctx, form.PushMirrorID, repo.ID)
if err != nil { if m == nil {
ctx.NotFound("", nil) ctx.NotFound("", nil)
return return
} }
if err = mirror_service.RemovePushMirrorRemote(ctx, m); err != nil { if err := mirror_service.RemovePushMirrorRemote(ctx, m); err != nil {
ctx.ServerError("RemovePushMirrorRemote", err) ctx.ServerError("RemovePushMirrorRemote", err)
return return
} }
if err = repo_model.DeletePushMirrors(ctx, repo_model.PushMirrorOptions{ID: m.ID, RepoID: m.RepoID}); err != nil { if err := repo_model.DeletePushMirrors(ctx, repo_model.PushMirrorOptions{ID: m.ID, RepoID: m.RepoID}); err != nil {
ctx.ServerError("DeletePushMirrorByID", err) ctx.ServerError("DeletePushMirrorByID", err)
return return
} }
@ -995,24 +995,3 @@ func handleSettingRemoteAddrError(ctx *context.Context, err error, form *forms.R
} }
ctx.RenderWithErr(ctx.Tr("repo.mirror_address_url_invalid"), tplSettingsOptions, form) ctx.RenderWithErr(ctx.Tr("repo.mirror_address_url_invalid"), tplSettingsOptions, form)
} }
func selectPushMirrorByForm(ctx *context.Context, form *forms.RepoSettingForm, repo *repo_model.Repository) (*repo_model.PushMirror, error) {
id, err := strconv.ParseInt(form.PushMirrorID, 10, 64)
if err != nil {
return nil, err
}
pushMirrors, _, err := repo_model.GetPushMirrorsByRepoID(ctx, repo.ID, db.ListOptions{})
if err != nil {
return nil, err
}
for _, m := range pushMirrors {
if m.ID == id {
m.Repo = repo
return m, nil
}
}
return nil, fmt.Errorf("PushMirror[%v] not associated to repository %v", id, repo)
}

View File

@ -122,7 +122,7 @@ type RepoSettingForm struct {
MirrorPassword string MirrorPassword string
LFS bool `form:"mirror_lfs"` LFS bool `form:"mirror_lfs"`
LFSEndpoint string `form:"mirror_lfs_endpoint"` LFSEndpoint string `form:"mirror_lfs_endpoint"`
PushMirrorID string PushMirrorID int64
PushMirrorAddress string PushMirrorAddress string
PushMirrorUsername string PushMirrorUsername string
PushMirrorPassword string PushMirrorPassword string

View File

@ -8,7 +8,6 @@ import (
"fmt" "fmt"
repo_model "code.gitea.io/gitea/models/repo" repo_model "code.gitea.io/gitea/models/repo"
"code.gitea.io/gitea/modules/graceful"
"code.gitea.io/gitea/modules/log" "code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/queue" "code.gitea.io/gitea/modules/queue"
"code.gitea.io/gitea/modules/setting" "code.gitea.io/gitea/modules/setting"
@ -119,14 +118,7 @@ func Update(ctx context.Context, pullLimit, pushLimit int) error {
return nil return nil
} }
func queueHandler(items ...*SyncRequest) []*SyncRequest {
for _, req := range items {
doMirrorSync(graceful.GetManager().ShutdownContext(), req)
}
return nil
}
// InitSyncMirrors initializes a go routine to sync the mirrors // InitSyncMirrors initializes a go routine to sync the mirrors
func InitSyncMirrors() { func InitSyncMirrors() {
StartSyncMirrors(queueHandler) StartSyncMirrors()
} }

View File

@ -28,12 +28,19 @@ type SyncRequest struct {
ReferenceID int64 // RepoID for pull mirror, MirrorID for push mirror ReferenceID int64 // RepoID for pull mirror, MirrorID for push mirror
} }
func queueHandler(items ...*SyncRequest) []*SyncRequest {
for _, req := range items {
doMirrorSync(graceful.GetManager().ShutdownContext(), req)
}
return nil
}
// StartSyncMirrors starts a go routine to sync the mirrors // StartSyncMirrors starts a go routine to sync the mirrors
func StartSyncMirrors(queueHandle func(data ...*SyncRequest) []*SyncRequest) { func StartSyncMirrors() {
if !setting.Mirror.Enabled { if !setting.Mirror.Enabled {
return return
} }
mirrorQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "mirror", queueHandle) mirrorQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "mirror", queueHandler)
if mirrorQueue == nil { if mirrorQueue == nil {
log.Fatal("Unable to create mirror queue") log.Fatal("Unable to create mirror queue")
} }

View File

@ -73,9 +73,12 @@ func TestDatabaseCollation(t *testing.T) {
t.Run("Convert tables to utf8mb4_bin", func(t *testing.T) { t.Run("Convert tables to utf8mb4_bin", func(t *testing.T) {
defer test.MockVariableValue(&setting.Database.CharsetCollation, "utf8mb4_bin")() defer test.MockVariableValue(&setting.Database.CharsetCollation, "utf8mb4_bin")()
assert.NoError(t, db.ConvertDatabaseTable())
r, err := db.CheckCollations(x) r, err := db.CheckCollations(x)
assert.NoError(t, err) assert.NoError(t, err)
assert.EqualValues(t, "utf8mb4_bin", r.ExpectedCollation)
assert.NoError(t, db.ConvertDatabaseTable())
r, err = db.CheckCollations(x)
assert.NoError(t, err)
assert.Equal(t, "utf8mb4_bin", r.DatabaseCollation) assert.Equal(t, "utf8mb4_bin", r.DatabaseCollation)
assert.True(t, r.CollationEquals(r.ExpectedCollation, r.DatabaseCollation)) assert.True(t, r.CollationEquals(r.ExpectedCollation, r.DatabaseCollation))
assert.Empty(t, r.InconsistentCollationColumns) assert.Empty(t, r.InconsistentCollationColumns)

View File

@ -9,7 +9,9 @@ import (
"net/http" "net/http"
"net/url" "net/url"
"strconv" "strconv"
"strings"
"testing" "testing"
"time"
"code.gitea.io/gitea/models/db" "code.gitea.io/gitea/models/db"
repo_model "code.gitea.io/gitea/models/repo" repo_model "code.gitea.io/gitea/models/repo"
@ -32,11 +34,10 @@ func TestMirrorPush(t *testing.T) {
} }
func testMirrorPush(t *testing.T, u *url.URL) { func testMirrorPush(t *testing.T, u *url.URL) {
defer tests.PrepareTestEnv(t)()
setting.Migrations.AllowLocalNetworks = true setting.Migrations.AllowLocalNetworks = true
assert.NoError(t, migrations.Init()) assert.NoError(t, migrations.Init())
_ = db.TruncateBeans(db.DefaultContext, &repo_model.PushMirror{})
user := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 2}) user := unittest.AssertExistsAndLoadBean(t, &user_model.User{ID: 2})
srcRepo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 1}) srcRepo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 1})
@ -45,9 +46,10 @@ func testMirrorPush(t *testing.T, u *url.URL) {
}) })
assert.NoError(t, err) assert.NoError(t, err)
ctx := NewAPITestContext(t, user.LowerName, srcRepo.Name) session := loginUser(t, user.Name)
doCreatePushMirror(ctx, fmt.Sprintf("%s%s/%s", u.String(), url.PathEscape(ctx.Username), url.PathEscape(mirrorRepo.Name)), user.LowerName, userPassword)(t) pushMirrorURL := fmt.Sprintf("%s%s/%s", u.String(), url.PathEscape(user.Name), url.PathEscape(mirrorRepo.Name))
testCreatePushMirror(t, session, user.Name, srcRepo.Name, pushMirrorURL, user.LowerName, userPassword, "0")
mirrors, _, err := repo_model.GetPushMirrorsByRepoID(db.DefaultContext, srcRepo.ID, db.ListOptions{}) mirrors, _, err := repo_model.GetPushMirrorsByRepoID(db.DefaultContext, srcRepo.ID, db.ListOptions{})
assert.NoError(t, err) assert.NoError(t, err)
@ -73,49 +75,81 @@ func testMirrorPush(t *testing.T, u *url.URL) {
assert.Equal(t, srcCommit.ID, mirrorCommit.ID) assert.Equal(t, srcCommit.ID, mirrorCommit.ID)
// Cleanup // Cleanup
doRemovePushMirror(ctx, fmt.Sprintf("%s%s/%s", u.String(), url.PathEscape(ctx.Username), url.PathEscape(mirrorRepo.Name)), user.LowerName, userPassword, int(mirrors[0].ID))(t) assert.True(t, doRemovePushMirror(t, session, user.Name, srcRepo.Name, mirrors[0].ID))
mirrors, _, err = repo_model.GetPushMirrorsByRepoID(db.DefaultContext, srcRepo.ID, db.ListOptions{}) mirrors, _, err = repo_model.GetPushMirrorsByRepoID(db.DefaultContext, srcRepo.ID, db.ListOptions{})
assert.NoError(t, err) assert.NoError(t, err)
assert.Len(t, mirrors, 0) assert.Len(t, mirrors, 0)
} }
func doCreatePushMirror(ctx APITestContext, address, username, password string) func(t *testing.T) { func testCreatePushMirror(t *testing.T, session *TestSession, owner, repo, address, username, password, interval string) {
return func(t *testing.T) { req := NewRequestWithValues(t, "POST", fmt.Sprintf("/%s/%s/settings", url.PathEscape(owner), url.PathEscape(repo)), map[string]string{
csrf := GetUserCSRFToken(t, ctx.Session) "_csrf": GetUserCSRFToken(t, session),
req := NewRequestWithValues(t, "POST", fmt.Sprintf("/%s/%s/settings", url.PathEscape(ctx.Username), url.PathEscape(ctx.Reponame)), map[string]string{
"_csrf": csrf,
"action": "push-mirror-add", "action": "push-mirror-add",
"push_mirror_address": address, "push_mirror_address": address,
"push_mirror_username": username, "push_mirror_username": username,
"push_mirror_password": password, "push_mirror_password": password,
"push_mirror_interval": "0", "push_mirror_interval": interval,
}) })
ctx.Session.MakeRequest(t, req, http.StatusSeeOther) session.MakeRequest(t, req, http.StatusSeeOther)
flashCookie := ctx.Session.GetCookie(gitea_context.CookieNameFlash) flashCookie := session.GetCookie(gitea_context.CookieNameFlash)
assert.NotNil(t, flashCookie) assert.NotNil(t, flashCookie)
assert.Contains(t, flashCookie.Value, "success") assert.Contains(t, flashCookie.Value, "success")
}
} }
func doRemovePushMirror(ctx APITestContext, address, username, password string, pushMirrorID int) func(t *testing.T) { func doRemovePushMirror(t *testing.T, session *TestSession, owner, repo string, pushMirrorID int64) bool {
return func(t *testing.T) { req := NewRequestWithValues(t, "POST", fmt.Sprintf("/%s/%s/settings", url.PathEscape(owner), url.PathEscape(repo)), map[string]string{
csrf := GetUserCSRFToken(t, ctx.Session) "_csrf": GetUserCSRFToken(t, session),
req := NewRequestWithValues(t, "POST", fmt.Sprintf("/%s/%s/settings", url.PathEscape(ctx.Username), url.PathEscape(ctx.Reponame)), map[string]string{
"_csrf": csrf,
"action": "push-mirror-remove", "action": "push-mirror-remove",
"push_mirror_id": strconv.Itoa(pushMirrorID), "push_mirror_id": strconv.FormatInt(pushMirrorID, 10),
"push_mirror_address": address,
"push_mirror_username": username,
"push_mirror_password": password,
"push_mirror_interval": "0",
}) })
ctx.Session.MakeRequest(t, req, http.StatusSeeOther) resp := session.MakeRequest(t, req, NoExpectedStatus)
flashCookie := session.GetCookie(gitea_context.CookieNameFlash)
flashCookie := ctx.Session.GetCookie(gitea_context.CookieNameFlash) return resp.Code == http.StatusSeeOther && flashCookie != nil && strings.Contains(flashCookie.Value, "success")
assert.NotNil(t, flashCookie) }
assert.Contains(t, flashCookie.Value, "success")
} func doUpdatePushMirror(t *testing.T, session *TestSession, owner, repo string, pushMirrorID int64, interval string) bool {
req := NewRequestWithValues(t, "POST", fmt.Sprintf("/%s/%s/settings", owner, repo), map[string]string{
"_csrf": GetUserCSRFToken(t, session),
"action": "push-mirror-update",
"push_mirror_id": strconv.FormatInt(pushMirrorID, 10),
"push_mirror_interval": interval,
"push_mirror_defer_sync": "true",
})
resp := session.MakeRequest(t, req, NoExpectedStatus)
return resp.Code == http.StatusSeeOther
}
func TestRepoSettingPushMirrorUpdate(t *testing.T) {
defer tests.PrepareTestEnv(t)()
setting.Migrations.AllowLocalNetworks = true
assert.NoError(t, migrations.Init())
session := loginUser(t, "user2")
repo2 := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 2})
testCreatePushMirror(t, session, "user2", "repo2", "https://127.0.0.1/user1/repo1.git", "", "", "24h")
pushMirrors, cnt, err := repo_model.GetPushMirrorsByRepoID(db.DefaultContext, repo2.ID, db.ListOptions{})
assert.NoError(t, err)
assert.EqualValues(t, 1, cnt)
assert.EqualValues(t, 24*time.Hour, pushMirrors[0].Interval)
repo2PushMirrorID := pushMirrors[0].ID
// update repo2 push mirror
assert.True(t, doUpdatePushMirror(t, session, "user2", "repo2", repo2PushMirrorID, "10m0s"))
pushMirror := unittest.AssertExistsAndLoadBean(t, &repo_model.PushMirror{ID: repo2PushMirrorID})
assert.EqualValues(t, 10*time.Minute, pushMirror.Interval)
// avoid updating repo2 push mirror from repo1
assert.False(t, doUpdatePushMirror(t, session, "user2", "repo1", repo2PushMirrorID, "20m0s"))
pushMirror = unittest.AssertExistsAndLoadBean(t, &repo_model.PushMirror{ID: repo2PushMirrorID})
assert.EqualValues(t, 10*time.Minute, pushMirror.Interval) // not changed
// avoid deleting repo2 push mirror from repo1
assert.False(t, doRemovePushMirror(t, session, "user2", "repo1", repo2PushMirrorID))
unittest.AssertExistsAndLoadBean(t, &repo_model.PushMirror{ID: repo2PushMirrorID})
// delete repo2 push mirror
assert.True(t, doRemovePushMirror(t, session, "user2", "repo2", repo2PushMirrorID))
unittest.AssertNotExistsBean(t, &repo_model.PushMirror{ID: repo2PushMirrorID})
} }