Compare commits

..

1 Commits

Author SHA1 Message Date
xxq250 3ab2e1c32c 导入仓库同步功能 2026-04-28 11:23:57 +08:00
16 changed files with 1910 additions and 117 deletions

View File

@ -1,68 +0,0 @@
version: 2
name: gitlink正式环境
description: ""
global:
concurrent: 1
workflow:
- ref: start
name: 开始
task: start
- ref: git_clone_0
name: git clone
on-failure: ignore
task: git_clone@1.2.6
input:
remote_url: '"https://gitlink.org.cn/Gitlink/gitea_hat.git"'
ref: '"refs/heads/master"'
commit_id: '""'
depth: 1
needs:
- start
- ref: end
name: 结束
task: end
needs:
- gitlink_ssh_cmd_0
- ref: gitlink_scp_resource_0
name: scp复制文件支持跳板机
on-failure: ignore
task: yystopf/gitlink_scp_resource@0.0.7
input:
ssh_private_key: ((ssh.siyao))
remote_host: '"10.9.117.109"'
remote_port: '"22"'
remote_user: '"root"'
remote_file: '"/root/gitea"'
local_file: golang_build_node_0.bin_dir
gateway_host: '"123.59.135.93"'
gateway_port: '"51123"'
gateway_user: '"pdl"'
temp_file: '"/home/pdl/gitea"'
needs:
- golang_build_node_0
- ref: golang_build_node_0
name: golang_build_node
on-failure: ignore
task: yystopf/golang_build_node@0.0.2
input:
workspace: git_clone_0.git_path
out_bin_name: '"gitea"'
goos: '"linux"'
goarch: '"amd64"'
needs:
- git_clone_0
- ref: gitlink_ssh_cmd_0
name: ssh执行命令支持跳板机
task: yystopf/gitlink_ssh_cmd@0.0.6
input:
ssh_private_key: ((ssh.siyao))
remote_host: '"10.9.117.109"'
remote_port: '"22"'
remote_user: '"root"'
gateway_host: '"123.59.135.93"'
gateway_port: '"51123"'
gateway_user: '"pdl"'
ssh_cmd: '"sh update.sh"'
needs:
- gitlink_scp_resource_0

View File

@ -1,40 +0,0 @@
version: 2
name: test
description: ""
global:
concurrent: 1
trigger:
webhook: gitlink@1.0.0
event:
- ref: push
ruleset-operator: AND
workflow:
- ref: start
name: 开始
task: start
- ref: git_clone_0
name: git clone
task: git_clone@1.2.9
input:
remote_url: '""'
ref: '"refs/heads/master"'
commit_id: '""'
depth: 1
needs:
- start
- ref: golong_build_122_0
name: golang build 1.22
task: dagege/golong_build_122@1.0.4
input:
workspace: git_clone_0.git_path
out_bin_name: catpaw
GOOS: '"linux"'
GOARCH: '"amd64"'
needs:
- git_clone_0
- ref: end
name: 结束
task: end
needs:
- golong_build_122_0

View File

@ -42,10 +42,6 @@
## 编译
```
sh build.sh
```
类Linux平台在项目根目录运行
```shell
# 编译Linux 64位可执行程序

3
go.sum
View File

@ -850,8 +850,9 @@ github.com/smartystreets/assertions v0.0.0-20190116191733-b6c0e53d7304/go.mod h1
github.com/smartystreets/assertions v1.1.1 h1:T/YLemO5Yp7KPzS+lVtu+WsHn8yoSwTfItdAd1r3cck=
github.com/smartystreets/assertions v1.1.1/go.mod h1:tcbTF8ujkAEcZ8TElKY+i30BzYlVhC/LOxJk7iOWnoo=
github.com/smartystreets/goconvey v0.0.0-20181108003508-044398e4856c/go.mod h1:XDJAKZRPZ1CvBcN2aX5YOUTYGHki24fSF0Iv48Ibg0s=
github.com/smartystreets/goconvey v0.0.0-20190731233626-505e41936337 h1:WN9BUFbdyOsSH/XohnWpXOlq9NBD5sGAB2FciQMUEe8=
github.com/smartystreets/goconvey v0.0.0-20190731233626-505e41936337/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
github.com/spf13/cobra v0.0.5/go.mod h1:3K3wKZymM7VvHMDS9+Akkh4K60UwM26emMESw8tLCHU=

106
http_test/sync-v2.http Normal file
View File

@ -0,0 +1,106 @@
### 1.pull-mirror-add
POST http://0.0.0.0:3003/api/hat/repos/xxq2506/sss2/mirror-sync-v2
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "mirror-sync",
"mirror_address": "http://127.0.0.1:3003/xxq250/aaabbbccc.git",
"mirror_username": "xxq250",
"mirror_password": "12345678",
"interval": "0"
}
### 1.pull-mirror-add
POST http://0.0.0.0:3003/api/hat/repos/xxq2506/sss2/mirror-sync-v2
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "push-mirror-add",
"mirror_address": "http://127.0.0.1:3003/xxq250/aaabbbccc.git",
"mirror_username": "xxq250",
"mirror_password": "12345678",
"interval": "0"
}
### 111.pull-mirror-add
POST http://0.0.0.0:3003/api/hat/repos/xxq250/bbb180/settings
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "pull-mirror-add",
"mirror_address": "https://www.gitlink.org.cn/xxq250/lfs-demo.git",
"mirror_username": "xxq250",
"mirror_password": "xxq123456789521",
"interval": "0"
}
### 2.pull-mirror-sync
POST http://0.0.0.0:3003/api/v1/repos/xxq250/bbb180/mirror-sync
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "mirror-sync"
}
### 2.pull-mirror-sync
POST http://0.0.0.0:3003/api/v1/repos/xxq2506/lfs-demo/mirror-sync
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "mirror-sync"
}
### 3.pull-mirror-remove
POST http://0.0.0.0:3003/api/hat/repos/xxq2506/aaabbbccc/mirror-sync-v2
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "pull-mirror-remove"
}
### 4.push-mirror-add
POST http://0.0.0.0:3003/api/hat/repos/xxq250/lfs-demo/settings
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "push-mirror-add",
"mirror_address": "https://www.gitlink.org.cn/xxq250/lfs-demo.git",
"mirror_username": "xxq250",
"mirror_password": "xxq123456789521",
"interval": "0"
}
### 5.push_mirrors-sync
POST http://0.0.0.0:3003/api/v1/repos/xxq250/lfs-demo/push_mirrors-sync
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
### 6.push-mirror-remove
POST http://0.0.0.0:3003/api/hat/repos/xxq250/lfs-demo/settings
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json

87
http_test/sync.http Normal file
View File

@ -0,0 +1,87 @@
### 1.pull-mirror-add
POST http://0.0.0.0:3003/api/hat/repos/xxq2506/lfs-demo/settings
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "pull-mirror-add",
"mirror_address": "http://0.0.0.0:3003/xxq250/lfs-demo.git",
"mirror_username": "xxq250",
"mirror_password": "xxq123456789521",
"interval": "0"
}
### 111.pull-mirror-add
POST http://0.0.0.0:3003/api/hat/repos/xxq250/bbb180/settings
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "pull-mirror-add",
"mirror_address": "https://www.gitlink.org.cn/xxq250/lfs-demo.git",
"mirror_username": "xxq250",
"mirror_password": "xxq123456789521",
"interval": "0"
}
### 2.pull-mirror-sync
POST http://0.0.0.0:3003/api/v1/repos/xxq250/bbb180/mirror-sync
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "mirror-sync"
}
### 2.pull-mirror-sync
POST http://0.0.0.0:3003/api/v1/repos/xxq2506/lfs-demo/mirror-sync
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "mirror-sync"
}
### 3.pull-mirror-remove
POST http://0.0.0.0:3003/api/hat/repos/xxq250/lfs-demo/settings
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
### 4.push-mirror-add
POST http://0.0.0.0:3003/api/hat/repos/xxq250/lfs-demo/settings
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
{
"action": "push-mirror-add",
"mirror_address": "https://www.gitlink.org.cn/xxq250/lfs-demo.git",
"mirror_username": "xxq250",
"mirror_password": "xxq123456789521",
"interval": "0"
}
### 5.push_mirrors-sync
POST http://0.0.0.0:3003/api/v1/repos/xxq250/lfs-demo/push_mirrors-sync
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json
### 6.push-mirror-remove
POST http://0.0.0.0:3003/api/hat/repos/xxq250/lfs-demo/settings
authorization: Basic eHhxMjUwOjEyMzQ1Njc4
accept: application/json
Content-Type: application/json

View File

@ -0,0 +1,204 @@
// Copyright 2023 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package repository
import (
"context"
"fmt"
"strings"
"code.gitea.io/gitea/models/db"
git_model "code.gitea.io/gitea/models/git"
repo_model "code.gitea.io/gitea/models/repo"
user_model "code.gitea.io/gitea/models/user"
"code.gitea.io/gitea/modules/container"
"code.gitea.io/gitea/modules/git"
gitea_git "code.gitea.io/gitea/modules/git"
"code.gitea.io/gitea/modules/log"
repo_module "code.gitea.io/gitea/modules/repository"
"code.gitea.io/gitea/modules/timeutil"
"code.gitea.io/gitea/modules/util"
repo_service "code.gitea.io/gitea/services/repository"
)
//// SyncRepoBranches synchronizes branch table with repository branches
//func SyncRepoBranches(ctx context.Context, repoID, doerID int64) (int64, error) {
// repo, err := repo_model.GetRepositoryByID(ctx, repoID)
// if err != nil {
// return 0, err
// }
//
// log.Debug("SyncRepoBranches: in Repo[%d:%s]", repo.ID, repo.FullName())
//
// gitRepo, err := git.OpenRepository(ctx, repo.RepoPath())
// if err != nil {
// log.Error("OpenRepository[%s]: %w", repo.RepoPath(), err)
// return 0, err
// }
// defer gitRepo.Close()
//
// return SyncRepoBranchesWithRepo(ctx, repo, gitRepo, doerID)
//}
func SyncRepoBranchesWithRepo(ctx context.Context, repo *repo_model.Repository, gitRepo *git.Repository, doerID int64, delOldBranches bool) (int64, error) {
allBranches := container.Set[string]{}
{
branches, _, err := gitRepo.GetBranchNames(0, 0)
if err != nil {
return 0, err
}
log.Trace("SyncRepoBranches[%s]: branches[%d]: %v", repo.FullName(), len(branches), branches)
for _, branch := range branches {
allBranches.Add(branch)
}
}
dbBranches := make(map[string]*git_model.Branch)
{
branches, err := git_model.FindBranches(ctx, git_model.FindBranchOptions{
ListOptions: db.ListOptions{
ListAll: true,
},
RepoID: repo.ID,
})
if err != nil {
return 0, err
}
for _, branch := range branches {
dbBranches[branch.Name] = branch
}
}
var toAdd []*git_model.Branch
var toUpdate []*git_model.Branch
var toRemove []int64
for branch := range allBranches {
dbb := dbBranches[branch]
commit, err := gitRepo.GetBranchCommit(branch)
if err != nil {
return 0, err
}
if dbb == nil {
toAdd = append(toAdd, &git_model.Branch{
RepoID: repo.ID,
Name: branch,
CommitID: commit.ID.String(),
CommitMessage: commit.Summary(),
PusherID: doerID,
CommitTime: timeutil.TimeStamp(commit.Committer.When.Unix()),
})
} else if commit.ID.String() != dbb.CommitID {
toUpdate = append(toUpdate, &git_model.Branch{
ID: dbb.ID,
RepoID: repo.ID,
Name: branch,
CommitID: commit.ID.String(),
CommitMessage: commit.Summary(),
PusherID: doerID,
CommitTime: timeutil.TimeStamp(commit.Committer.When.Unix()),
})
}
}
for _, dbBranch := range dbBranches {
if !allBranches.Contains(dbBranch.Name) && !dbBranch.IsDeleted {
toRemove = append(toRemove, dbBranch.ID)
}
}
log.Trace("SyncRepoBranches[%s]: toAdd: %v, toUpdate: %v, toRemove: %v", repo.FullName(), toAdd, toUpdate, toRemove)
if len(toAdd) == 0 && len(toRemove) == 0 && len(toUpdate) == 0 {
return int64(len(allBranches)), nil
}
if err := db.WithTx(ctx, func(ctx context.Context) error {
if len(toAdd) > 0 {
if err := git_model.AddBranches(ctx, toAdd); err != nil {
return err
}
}
for _, b := range toUpdate {
if _, err := db.GetEngine(ctx).ID(b.ID).
Cols("commit_id, commit_message, pusher_id, commit_time, is_deleted").
Update(b); err != nil {
return err
}
}
if len(toRemove) > 0 {
if err := git_model.DeleteBranches(ctx, repo.ID, doerID, toRemove); err != nil {
return err
}
// default DeleteBranches after restore
if !delOldBranches {
_, err := RestoreDeleteBranches(ctx, repo.ID)
if err != nil {
log.Error("SyncMirrors RestoreDeleteBranches error", err)
return err
}
}
}
return nil
}); err != nil {
return 0, err
}
return int64(len(allBranches)), nil
}
func RestoreDeleteBranches(ctx context.Context, repoID int64) (int64, error) {
branches, err := git_model.FindBranches(ctx, git_model.FindBranchOptions{
ListOptions: db.ListOptions{
ListAll: true,
},
IsDeletedBranch: util.OptionalBoolTrue,
RepoID: repoID,
})
if err != nil {
log.Error("GetBranches", err.Error())
return 0, err
}
log.Info("FindDeleteBranches len(branches)=========== %s", len(branches))
for i := range branches {
deletedBranch := branches[i]
repo, err := repo_model.GetRepositoryByID(ctx, repoID)
if err != nil {
return 0, err
}
doer, err := user_model.GetUserByID(ctx, repo.OwnerID)
if err != nil {
return 0, err
}
repo.Owner = doer
if err := gitea_git.Push(ctx, repo.RepoPath(), gitea_git.PushOptions{
Remote: repo.RepoPath(),
Branch: fmt.Sprintf("%s:%s%s", deletedBranch.CommitID, gitea_git.BranchPrefix, deletedBranch.Name),
Env: repo_module.PushingEnvironment(repo.Owner, repo),
}); err != nil {
if strings.Contains(err.Error(), "already exists") {
log.Error("RestoreBranch: Can't restore branch '%s', since one with same name already exist", deletedBranch.Name)
return 0, err
}
log.Error("RestoreBranch: CreateBranch: %v", err.Error())
return 0, err
}
if err := repo_service.PushUpdate(
&repo_module.PushUpdateOptions{
RefFullName: gitea_git.RefNameFromBranch(deletedBranch.Name),
OldCommitID: gitea_git.EmptySHA,
NewCommitID: deletedBranch.CommitID,
PusherID: repo.Owner.ID,
PusherName: repo.Owner.Name,
RepoUserName: repo.Owner.Name,
RepoName: repo.Name,
}); err != nil {
log.Error("RestoreBranch: Update: %v", err.Error())
return 0, nil
}
}
return 0, nil
}

View File

@ -253,6 +253,7 @@ func Routers() *web.Route {
}, context.ReferencesGitRepo(), reqRepoReader(unit.TypeCode))
m.Get("/blame", context.ReferencesGitRepo(), repo.GetRepoRefBlame)
m.Get("/code_stats", context.ReferencesGitRepo(), repo.ListCodeStats)
m.Post("/mirror-sync-v2", bind(repo.RepoSyncFormNew{}), repo.MirrorSyncV2)
}, repoAssignment())
})
m.Group("/users", func() {

View File

@ -0,0 +1,387 @@
package repo
import (
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"time"
"code.gitea.io/gitea/models"
repo_model "code.gitea.io/gitea/models/repo"
"code.gitea.io/gitea/modules/context"
"code.gitea.io/gitea/modules/git"
"code.gitea.io/gitea/modules/lfs"
"code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/setting"
"code.gitea.io/gitea/modules/timeutil"
"code.gitea.io/gitea/modules/util"
"code.gitea.io/gitea/modules/web"
"code.gitea.io/gitea/services/forms"
"code.gitea.io/gitea/services/migrations"
hat_mirror_service "code.gitlink.org.cn/Gitlink/gitea_hat.git/services/mirror"
)
// RepoSettingForm form for changing repository settings
type RepoSyncFormNew struct {
//RepoName string `binding:"Required;AlphaDashDot;MaxSize(100)"`
//Description string `binding:"MaxSize(2048)"`
//Website string `binding:"ValidUrl;MaxSize(1024)"`
Action string `json:"action"`
Interval string `json:"interval"`
MirrorAddress string `json:"mirror_address"`
MirrorUsername string `json:"mirror_username"`
MirrorPassword string `json:"mirror_password"`
LFS bool `json:"mirror_lfs"`
LFSEndpoint string `json:"mirror_lfs_endpoint"`
DelOldBranches bool `json:"del_old_branches"`
//PushMirrorID string `json:"push_mirror_id"`
//PushMirrorAddress string `json:"push_mirror_address"`
//PushMirrorUsername string `json:"push_mirror_username"`
//PushMirrorPassword string `json:"push_mirror_password"`
MirrorSyncOnCommit bool `json:"mirror_sync_on_commit"`
//PushMirrorInterval string `json:"push_mirror_interval"`
//Private bool `json:"private"`
//Template bool `json:"Template"`
EnablePrune bool `json:"enable_prune"`
}
func PullMirrorAdd(ctx *context.APIContext, form *RepoSyncFormNew, repo *repo_model.Repository) {
mirrorModel := repo_model.Mirror{
RepoID: repo.ID,
Interval: setting.Mirror.DefaultInterval,
EnablePrune: true,
NextUpdateUnix: timeutil.TimeStampNow().AddDuration(setting.Mirror.DefaultInterval),
LFS: false,
RemoteAddress: form.MirrorAddress,
}
parsedInterval, err := time.ParseDuration(form.Interval)
if err != nil {
log.Error("Failed to set Interval: %v", err)
ctx.Error(http.StatusInternalServerError, "Err_Interval", err)
return
}
if parsedInterval == 0 {
mirrorModel.Interval = 0
mirrorModel.NextUpdateUnix = 0
} else if parsedInterval < setting.Mirror.MinInterval {
err := fmt.Errorf("interval %s is set below Minimum Interval of %s", parsedInterval, setting.Mirror.MinInterval)
log.Error("Interval: %s is too frequent", form.Interval)
ctx.Error(http.StatusInternalServerError, "Err_Interval too max", err)
return
} else {
mirrorModel.Interval = parsedInterval
mirrorModel.NextUpdateUnix = timeutil.TimeStampNow().AddDuration(parsedInterval)
}
if err = repo_model.InsertMirror(ctx, &mirrorModel); err != nil {
ctx.Error(http.StatusInternalServerError, "mirrorAdd_err", err)
return
}
repoPath := repo_model.RepoPath(repo.OwnerName, repo.Name)
// this is necessary for sync local tags from remote
configName := fmt.Sprintf("remote.%s.fetch", mirrorModel.GetRemoteName())
_, _, err = git.NewCommand(ctx, "remote", "rm").AddDynamicArguments(mirrorModel.GetRemoteName()).RunStdString(&git.RunOpts{Dir: repoPath})
cmd := git.NewCommand(ctx, "remote", "add").AddDynamicArguments(mirrorModel.GetRemoteName()).AddArguments("--mirror=fetch").AddDynamicArguments(mirrorModel.RemoteAddress)
if strings.Contains(mirrorModel.RemoteAddress, "://") && strings.Contains(mirrorModel.RemoteAddress, "@") {
cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=fetch %s [repo_path: %s]", mirrorModel.GetRemoteName(), util.SanitizeCredentialURLs(mirrorModel.RemoteAddress), repoPath))
} else {
cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=fetch %s [repo_path: %s]", mirrorModel.GetRemoteName(), mirrorModel.RemoteAddress, repoPath))
}
_, _, err = cmd.RunStdString(&git.RunOpts{Dir: repoPath})
if err != nil && !strings.HasPrefix(err.Error(), "exit status 128 - fatal: No such remote ") {
ctx.Error(http.StatusInternalServerError, "error in MigrateRepositoryGitData(git config --add <remote> +refs/tags/*:refs/tags/*): %w", err)
return
}
if stdout, _, err := git.NewCommand(ctx, "config").AddOptionValues("--add", configName, `+refs/tags/*:refs/tags/*`).
RunStdString(&git.RunOpts{Dir: repoPath}); err != nil {
log.Error("MigrateRepositoryGitData(git config --add <remote> +refs/tags/*:refs/tags/*) in %v: Stdout: %s\nError: %v", repo, stdout, err)
ctx.Error(http.StatusInternalServerError, "error in MigrateRepositoryGitData(git config --add <remote> +refs/tags/*:refs/tags/*): %w", err)
return
}
}
func PullMirrorUpdate(ctx *context.APIContext, form *RepoSyncFormNew, repo *repo_model.Repository) {
pullMirror, err := repo_model.GetMirrorByRepoID(ctx, ctx.Repo.Repository.ID)
if err == repo_model.ErrMirrorNotExist {
ctx.NotFound("", nil)
return
}
if err != nil {
ctx.Error(http.StatusInternalServerError, "GetMirrorByRepoID", err)
return
}
interval, err := time.ParseDuration(form.Interval)
if err != nil || (interval != 0 && interval < setting.Mirror.MinInterval) {
ctx.Data["Err_Interval"] = true
ctx.Error(http.StatusInternalServerError, "Err_Interval", err)
return
}
pullMirror.EnablePrune = form.EnablePrune
pullMirror.Interval = interval
pullMirror.ScheduleNextUpdate()
if err := repo_model.UpdateMirror(ctx, pullMirror); err != nil {
ctx.Error(http.StatusInternalServerError, "UpdateMirror", err)
return
}
u, err := git.GetRemoteURL(ctx, ctx.Repo.Repository.RepoPath(), pullMirror.GetRemoteName())
if err != nil {
ctx.Data["Err_MirrorAddress"] = true
ctx.Error(http.StatusInternalServerError, "repo.Err_MirrorAddress", err)
return
}
if u.User != nil && form.MirrorPassword == "" && form.MirrorUsername == u.User.Username() {
form.MirrorPassword, _ = u.User.Password()
}
address, err := forms.ParseRemoteAddr(form.MirrorAddress, form.MirrorUsername, form.MirrorPassword)
if err == nil {
err = migrations.IsMigrateURLAllowed(address, ctx.Doer)
}
if err != nil {
ctx.Data["Err_MirrorAddress"] = true
ctx.Error(http.StatusInternalServerError, "Err_MirrorAddress", err)
return
}
if err := hat_mirror_service.UpdateAddress(ctx, pullMirror, address); err != nil {
ctx.Error(http.StatusInternalServerError, "UpdateAddress", err)
return
}
remoteAddress, err := util.SanitizeURL(form.MirrorAddress)
if err != nil {
ctx.Error(http.StatusInternalServerError, "SanitizeURL", err)
return
}
pullMirror.RemoteAddress = remoteAddress
form.LFS = form.LFS && setting.LFS.StartServer
if len(form.LFSEndpoint) > 0 {
ep := lfs.DetermineEndpoint("", form.LFSEndpoint)
if ep == nil {
ctx.Data["Err_LFSEndpoint"] = true
ctx.Error(http.StatusInternalServerError, "Err_LFSEndpoint", err)
return
}
err = migrations.IsMigrateURLAllowed(ep.String(), ctx.Doer)
if err != nil {
ctx.Data["Err_LFSEndpoint"] = true
ctx.Error(http.StatusInternalServerError, "Err_LFSEndpoint", err)
return
}
}
pullMirror.LFS = form.LFS
pullMirror.LFSEndpoint = form.LFSEndpoint
if err := repo_model.UpdateMirror(ctx, pullMirror); err != nil {
ctx.Error(http.StatusInternalServerError, "UpdateMirror", err)
return
}
ctx.JSON(http.StatusOK, map[string]any{"ok": true})
}
func PullMirrorRemove(ctx *context.APIContext, form *RepoSyncFormNew, repo *repo_model.Repository) {
_, err1 := repo_model.GetMirrorByRepoID(ctx, ctx.Repo.Repository.ID)
if err1 != nil {
ctx.Error(http.StatusInternalServerError, "pullMirror already exists, repoId:", strconv.FormatInt(ctx.Repo.Repository.ID, 10))
return
}
if err := repo_model.DeleteMirrorByRepoID(ctx, ctx.Repo.Repository.ID); err != nil {
ctx.Error(http.StatusInternalServerError, "DeleteMirrorByRepoID", err)
return
}
repoPath := repo_model.RepoPath(repo.OwnerName, repo.Name)
_, _, err := git.NewCommand(ctx, "remote", "rm", "origin").RunStdString(&git.RunOpts{Dir: repoPath})
if err != nil && !strings.HasPrefix(err.Error(), "exit status 128 - fatal: No such remote ") {
ctx.Error(http.StatusInternalServerError, "DeleteMirrorConfig", err)
return
}
ctx.JSON(http.StatusOK, map[string]any{"ok": true})
}
func PushMirrorUpdate(ctx *context.APIContext, form *RepoSyncFormNew, repo *repo_model.Repository) {
interval, err := time.ParseDuration(form.Interval)
if err != nil || (interval != 0 && interval < setting.Mirror.MinInterval) {
ctx.Error(http.StatusInternalServerError, "mirror_interval_invalid", err)
return
}
pushMirror, err := repo_model.GetPushMirror(ctx, repo_model.PushMirrorOptions{RepoID: ctx.Repo.Repository.ID})
if err != nil {
ctx.Error(http.StatusInternalServerError, "GetPushMirror", err)
return
}
m := &repo_model.PushMirror{
ID: pushMirror.ID,
Interval: interval,
}
if err := repo_model.UpdatePushMirrorInterval(ctx, m); err != nil {
ctx.Error(http.StatusInternalServerError, "UpdatePushMirror Interval", err)
return
}
// Background why we are adding it to Queue
// 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.
// So, there are updates within the given interval, it is necessary to update the queue accordingly.
hat_mirror_service.AddPushMirrorToQueue(m.ID)
ctx.JSON(http.StatusOK, map[string]any{"ok": true})
}
func PushMirrorAdd(ctx *context.APIContext, form *RepoSyncFormNew, repo *repo_model.Repository) {
pushMirror, err := repo_model.GetPushMirror(ctx, repo_model.PushMirrorOptions{RepoID: ctx.Repo.Repository.ID})
if err == nil && pushMirror != nil {
ctx.Error(http.StatusInternalServerError, "pushMirror already exists, repoId:", strconv.FormatInt(ctx.Repo.Repository.ID, 10))
return
}
// This section doesn't require repo_name/RepoName to be set in the form, don't show it
// as an error on the UI for this action
ctx.Data["Err_RepoName"] = nil
interval, err := time.ParseDuration(form.Interval)
if err != nil || (interval != 0 && interval < setting.Mirror.MinInterval) {
ctx.Data["Err_PushMirror Interval"] = true
ctx.Error(http.StatusInternalServerError, "Err_PushMirror Interval", err)
return
}
address, err := forms.ParseRemoteAddr(form.MirrorAddress, form.MirrorUsername, form.MirrorPassword)
if err == nil {
err = migrations.IsMigrateURLAllowed(address, ctx.Doer)
}
if err != nil {
ctx.Data["Err_PushMirrorAddress"] = true
ctx.Error(http.StatusInternalServerError, "Err_PushMirrorAddress", err)
return
}
remoteSuffix, err := util.CryptoRandomString(10)
if err != nil {
ctx.Error(http.StatusInternalServerError, "RandomString", err)
return
}
remoteAddress, err := util.SanitizeURL(form.MirrorAddress)
if err != nil {
ctx.Error(http.StatusInternalServerError, "SanitizeURL", err)
return
}
m := &repo_model.PushMirror{
RepoID: repo.ID,
Repo: repo,
RemoteName: fmt.Sprintf("remote_mirror_%s", remoteSuffix),
SyncOnCommit: form.MirrorSyncOnCommit,
Interval: interval,
RemoteAddress: remoteAddress,
}
if err := repo_model.InsertPushMirror(ctx, m); err != nil {
ctx.Error(http.StatusInternalServerError, "InsertPushMirror", err)
return
}
if err := hat_mirror_service.AddPushMirrorRemote(ctx, m, address); err != nil {
if err := repo_model.DeletePushMirrors(ctx, repo_model.PushMirrorOptions{ID: m.ID, RepoID: m.RepoID}); err != nil {
log.Error("DeletePushMirrors %v", err)
}
ctx.Error(http.StatusInternalServerError, "AddPushMirrorRemote", err)
return
}
ctx.JSON(http.StatusOK, map[string]any{"ok": true})
}
func PushMirrorRemove(ctx *context.APIContext, form *RepoSyncFormNew, repo *repo_model.Repository) {
pushMirror, err := repo_model.GetPushMirror(ctx, repo_model.PushMirrorOptions{RepoID: ctx.Repo.Repository.ID})
if err != nil {
ctx.Error(http.StatusInternalServerError, "GetPushMirror", err)
return
}
if err = hat_mirror_service.RemovePushMirrorRemote(ctx, pushMirror); err != nil {
ctx.Error(http.StatusInternalServerError, "RemovePushMirrorRemote", err)
return
}
if err = repo_model.DeletePushMirrors(ctx, repo_model.PushMirrorOptions{ID: pushMirror.ID, RepoID: pushMirror.RepoID}); err != nil {
ctx.Error(http.StatusInternalServerError, "DeletePushMirrorByID", err)
return
}
ctx.JSON(http.StatusOK, map[string]any{"ok": true})
}
func MirrorSyncV2(ctx *context.APIContext) {
form := web.GetForm(ctx).(*RepoSyncFormNew)
repo := ctx.Repo.Repository
if form.Action == "mirror-sync" || form.Action == "push-mirror-update" || form.Action == "push-mirror-update" {
remoteAddr, err := forms.ParseRemoteAddr(form.MirrorAddress, form.MirrorUsername, form.MirrorPassword)
if err == nil {
err = migrations.IsMigrateURLAllowed(remoteAddr, ctx.Doer)
}
if err != nil {
handleRemoteAddrError(ctx, err)
return
}
form.MirrorAddress = remoteAddr
}
switch form.Action {
case "mirror-sync":
_, err := repo_model.GetMirrorByRepoID(ctx, ctx.Repo.Repository.ID)
// 检测pullMirror不存在,增加
if err != nil {
if errors.Is(err, repo_model.ErrMirrorNotExist) {
PullMirrorAdd(ctx, form, repo)
} else {
ctx.Error(http.StatusInternalServerError, "MirrorSync", err)
return
}
}
// 开始同步加入队列
hat_mirror_service.AddPullMirrorToQueue(repo.ID, form.DelOldBranches)
ctx.JSON(http.StatusOK, map[string]any{"ok": true})
case "pull-mirror-update":
PullMirrorUpdate(ctx, form, repo)
case "pull-mirror-remove":
PullMirrorRemove(ctx, form, repo)
case "push-mirror-update":
PushMirrorUpdate(ctx, form, repo)
case "push-mirror-add":
PushMirrorAdd(ctx, form, repo)
case "push-mirror-remove":
PushMirrorRemove(ctx, form, repo)
default:
ctx.NotFound("", nil)
}
}
func handleRemoteAddrError(ctx *context.APIContext, err error) {
if models.IsErrInvalidCloneAddr(err) {
addrErr := err.(*models.ErrInvalidCloneAddr)
switch {
case addrErr.IsURLError:
ctx.Error(http.StatusUnprocessableEntity, "", err)
case addrErr.IsPermissionDenied:
if addrErr.LocalPath {
ctx.Error(http.StatusUnprocessableEntity, "", "You are not allowed to import local repositories.")
} else {
ctx.Error(http.StatusUnprocessableEntity, "", "You can not import from disallowed hosts.")
}
case addrErr.IsInvalidPath:
ctx.Error(http.StatusUnprocessableEntity, "", "Invalid local path, it does not exist or not a directory.")
default:
ctx.Error(http.StatusInternalServerError, "ParseRemoteAddr", "Unknown error type (ErrInvalidCloneAddr): "+err.Error())
}
} else {
ctx.Error(http.StatusInternalServerError, "ParseRemoteAddr", err)
}
}

View File

@ -283,9 +283,9 @@ func registerRoutes(m *web.Route) {
reqSignOut := verifyAuthWithOptions(&common.VerifyOptions{SignOutRequired: true})
// TODO: rename them to "optSignIn", which means that the "sign-in" could be optional, depends on the VerifyOptions (RequireSignInView)
//ignSignIn := verifyAuthWithOptions(&common.VerifyOptions{SignInRequired: setting.Service.RequireSignInView})
ignSignIn := verifyAuthWithOptions(&common.VerifyOptions{SignInRequired: true})
//ignExploreSignIn := verifyAuthWithOptions(&common.VerifyOptions{SignInRequired: setting.Service.RequireSignInView || setting.Service.Explore.RequireSigninView})
ignExploreSignIn := verifyAuthWithOptions(&common.VerifyOptions{SignInRequired: true})
ignSignIn := verifyAuthWithOptions(&common.VerifyOptions{SignInRequired: setting.Service.Explore.RequireSigninView})
ignExploreSignIn := verifyAuthWithOptions(&common.VerifyOptions{SignInRequired: setting.Service.RequireSignInView || setting.Service.Explore.RequireSigninView})
//ignExploreSignIn := verifyAuthWithOptions(&common.VerifyOptions{SignInRequired: true})
validation.AddBindingRules()

View File

@ -21,6 +21,7 @@ import (
"code.gitlink.org.cn/Gitlink/gitea_hat.git/models/migrations"
api_hat "code.gitlink.org.cn/Gitlink/gitea_hat.git/routers/hat"
web_routers "code.gitlink.org.cn/Gitlink/gitea_hat.git/routers/hat/web"
hat_mirror_service "code.gitlink.org.cn/Gitlink/gitea_hat.git/services/mirror"
hat_pull_service "code.gitlink.org.cn/Gitlink/gitea_hat.git/services/pull"
)
@ -45,7 +46,7 @@ func mustInitCtx(ctx context.Context, fn func(ctx context.Context) error) {
func GlobalInitInstalled(ctx context.Context) {
mustInitCtx(ctx, InitDBEngine)
mustInit(hat_pull_service.Init)
mustInit(hat_mirror_service.InitSyncMirrors)
}
func InitDBEngine(ctx context.Context) (err error) {

133
services/mirror/mirror.go Normal file
View File

@ -0,0 +1,133 @@
// Copyright 2019 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package mirror
import (
"context"
"fmt"
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/queue"
"code.gitea.io/gitea/modules/setting"
)
// doMirrorSync causes this request to mirror itself
func doMirrorSync(ctx context.Context, req *SyncRequest) {
if req.ReferenceID == 0 {
log.Warn("Skipping mirror sync request, no mirror ID was specified")
return
}
switch req.Type {
case PushMirrorType:
_ = SyncPushMirror(ctx, req.ReferenceID)
case PullMirrorType:
_ = SyncPullMirror(ctx, req.ReferenceID, req.DelOldBranches)
default:
log.Error("Unknown Request type in queue: %v for MirrorID[%d]", req.Type, req.ReferenceID)
}
}
var errLimit = fmt.Errorf("reached limit")
// Update checks and updates mirror repositories.
func Update(ctx context.Context, pullLimit, pushLimit int) error {
if !setting.Mirror.Enabled {
log.Warn("Mirror feature disabled, but cron job enabled: skip update")
return nil
}
log.Trace("Doing: Update")
handler := func(idx int, bean any) error {
var repo *repo_model.Repository
var mirrorType SyncType
var referenceID int64
if m, ok := bean.(*repo_model.Mirror); ok {
if m.GetRepository(ctx) == nil {
log.Error("Disconnected mirror found: %d", m.ID)
return nil
}
repo = m.Repo
mirrorType = PullMirrorType
referenceID = m.RepoID
} else if m, ok := bean.(*repo_model.PushMirror); ok {
if m.GetRepository() == nil {
log.Error("Disconnected push-mirror found: %d", m.ID)
return nil
}
repo = m.Repo
mirrorType = PushMirrorType
referenceID = m.ID
} else {
log.Error("Unknown bean: %v", bean)
return nil
}
// Check we've not been cancelled
select {
case <-ctx.Done():
return fmt.Errorf("aborted")
default:
}
// Push to the Queue
if err := PushToQueue(mirrorType, referenceID, (mirrorType == PullMirrorType)); err != nil {
if err == queue.ErrAlreadyInQueue {
if mirrorType == PushMirrorType {
log.Trace("PushMirrors for %-v already queued for sync", repo)
} else {
log.Trace("PullMirrors for %-v already queued for sync", repo)
}
return nil
}
return err
}
return nil
}
pullMirrorsRequested := 0
if pullLimit != 0 {
if err := repo_model.MirrorsIterate(ctx, pullLimit, func(idx int, bean any) error {
if err := handler(idx, bean); err != nil {
return err
}
pullMirrorsRequested++
return nil
}); err != nil && err != errLimit {
log.Error("MirrorsIterate: %v", err)
return err
}
}
pushMirrorsRequested := 0
if pushLimit != 0 {
if err := repo_model.PushMirrorsIterate(ctx, pushLimit, func(idx int, bean any) error {
if err := handler(idx, bean); err != nil {
return err
}
pushMirrorsRequested++
return nil
}); err != nil && err != errLimit {
log.Error("PushMirrorsIterate: %v", err)
return err
}
}
log.Trace("Finished: Update: %d pull mirrors and %d push mirrors queued", pullMirrorsRequested, pushMirrorsRequested)
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
func InitSyncMirrors() error {
StartSyncMirrors(queueHandler)
return nil
}

View File

@ -0,0 +1,615 @@
// Copyright 2021 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package mirror
import (
"context"
"fmt"
"strings"
"time"
"code.gitea.io/gitea/models/db"
repo_model "code.gitea.io/gitea/models/repo"
system_model "code.gitea.io/gitea/models/system"
"code.gitea.io/gitea/modules/cache"
"code.gitea.io/gitea/modules/git"
"code.gitea.io/gitea/modules/lfs"
"code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/process"
"code.gitea.io/gitea/modules/proxy"
repo_module "code.gitea.io/gitea/modules/repository"
"code.gitea.io/gitea/modules/setting"
"code.gitea.io/gitea/modules/timeutil"
"code.gitea.io/gitea/modules/util"
notify_service "code.gitea.io/gitea/services/notify"
hat_repo_module "code.gitlink.org.cn/Gitlink/gitea_hat.git/modules/repository"
)
// gitShortEmptySha Git short empty SHA
const gitShortEmptySha = "0000000"
// UpdateAddress writes new address to Git repository and database
func UpdateAddress(ctx context.Context, m *repo_model.Mirror, addr string) error {
remoteName := m.GetRemoteName()
repoPath := m.GetRepository(ctx).RepoPath()
// Remove old remote
_, _, err := git.NewCommand(ctx, "remote", "rm").AddDynamicArguments(remoteName).RunStdString(&git.RunOpts{Dir: repoPath})
if err != nil && !strings.HasPrefix(err.Error(), "exit status 128 - fatal: No such remote ") {
return err
}
cmd := git.NewCommand(ctx, "remote", "add").AddDynamicArguments(remoteName).AddArguments("--mirror=fetch").AddDynamicArguments(addr)
if strings.Contains(addr, "://") && strings.Contains(addr, "@") {
cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=fetch %s [repo_path: %s]", remoteName, util.SanitizeCredentialURLs(addr), repoPath))
} else {
cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=fetch %s [repo_path: %s]", remoteName, addr, repoPath))
}
_, _, err = cmd.RunStdString(&git.RunOpts{Dir: repoPath})
if err != nil && !strings.HasPrefix(err.Error(), "exit status 128 - fatal: No such remote ") {
return err
}
if m.Repo.HasWiki() {
wikiPath := m.Repo.WikiPath()
wikiRemotePath := repo_module.WikiRemoteURL(ctx, addr)
// Remove old remote of wiki
_, _, err = git.NewCommand(ctx, "remote", "rm").AddDynamicArguments(remoteName).RunStdString(&git.RunOpts{Dir: wikiPath})
if err != nil && !strings.HasPrefix(err.Error(), "exit status 128 - fatal: No such remote ") {
return err
}
cmd = git.NewCommand(ctx, "remote", "add").AddDynamicArguments(remoteName).AddArguments("--mirror=fetch").AddDynamicArguments(wikiRemotePath)
if strings.Contains(wikiRemotePath, "://") && strings.Contains(wikiRemotePath, "@") {
cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=fetch %s [repo_path: %s]", remoteName, util.SanitizeCredentialURLs(wikiRemotePath), wikiPath))
} else {
cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=fetch %s [repo_path: %s]", remoteName, wikiRemotePath, wikiPath))
}
_, _, err = cmd.RunStdString(&git.RunOpts{Dir: wikiPath})
if err != nil && !strings.HasPrefix(err.Error(), "exit status 128 - fatal: No such remote ") {
return err
}
}
m.Repo.OriginalURL = addr
return repo_model.UpdateRepositoryCols(ctx, m.Repo, "original_url")
}
// mirrorSyncResult contains information of a updated reference.
// If the oldCommitID is "0000000", it means a new reference, the value of newCommitID is empty.
// If the newCommitID is "0000000", it means the reference is deleted, the value of oldCommitID is empty.
type mirrorSyncResult struct {
refName git.RefName
oldCommitID string
newCommitID string
}
// parseRemoteUpdateOutput detects create, update and delete operations of references from upstream.
// possible output example:
/*
// * [new tag] v0.1.8 -> v0.1.8
// * [new branch] master -> origin/master
// - [deleted] (none) -> origin/test // delete a branch
// - [deleted] (none) -> 1 // delete a tag
// 957a993..a87ba5f test -> origin/test
// + f895a1e...957a993 test -> origin/test (forced update)
*/
// TODO: return whether it's a force update
func parseRemoteUpdateOutput(output, remoteName string) []*mirrorSyncResult {
results := make([]*mirrorSyncResult, 0, 3)
lines := strings.Split(output, "\n")
for i := range lines {
// Make sure reference name is presented before continue
idx := strings.Index(lines[i], "-> ")
if idx == -1 {
continue
}
refName := strings.TrimSpace(lines[i][idx+3:])
switch {
case strings.HasPrefix(lines[i], " * [new tag]"): // new tag
results = append(results, &mirrorSyncResult{
refName: git.RefNameFromTag(refName),
oldCommitID: gitShortEmptySha,
})
case strings.HasPrefix(lines[i], " * [new branch]"): // new branch
refName = strings.TrimPrefix(refName, remoteName+"/")
results = append(results, &mirrorSyncResult{
refName: git.RefNameFromBranch(refName),
oldCommitID: gitShortEmptySha,
})
case strings.HasPrefix(lines[i], " - "): // Delete reference
isTag := !strings.HasPrefix(refName, remoteName+"/")
var refFullName git.RefName
if isTag {
refFullName = git.RefNameFromTag(refName)
} else {
refFullName = git.RefNameFromBranch(strings.TrimPrefix(refName, remoteName+"/"))
}
results = append(results, &mirrorSyncResult{
refName: refFullName,
newCommitID: gitShortEmptySha,
})
case strings.HasPrefix(lines[i], " + "): // Force update
if idx := strings.Index(refName, " "); idx > -1 {
refName = refName[:idx]
}
delimIdx := strings.Index(lines[i][3:], " ")
if delimIdx == -1 {
log.Error("SHA delimiter not found: %q", lines[i])
continue
}
shas := strings.Split(lines[i][3:delimIdx+3], "...")
if len(shas) != 2 {
log.Error("Expect two SHAs but not what found: %q", lines[i])
continue
}
results = append(results, &mirrorSyncResult{
refName: git.RefNameFromBranch(strings.TrimPrefix(refName, remoteName+"/")),
oldCommitID: shas[0],
newCommitID: shas[1],
})
case strings.HasPrefix(lines[i], " "): // New commits of a reference
delimIdx := strings.Index(lines[i][3:], " ")
if delimIdx == -1 {
log.Error("SHA delimiter not found: %q", lines[i])
continue
}
shas := strings.Split(lines[i][3:delimIdx+3], "..")
if len(shas) != 2 {
log.Error("Expect two SHAs but not what found: %q", lines[i])
continue
}
results = append(results, &mirrorSyncResult{
refName: git.RefNameFromBranch(strings.TrimPrefix(refName, remoteName+"/")),
oldCommitID: shas[0],
newCommitID: shas[1],
})
default:
log.Warn("parseRemoteUpdateOutput: unexpected update line %q", lines[i])
}
}
return results
}
func pruneBrokenReferences(ctx context.Context,
m *repo_model.Mirror,
repoPath string,
timeout time.Duration,
stdoutBuilder, stderrBuilder *strings.Builder,
isWiki bool,
) error {
wiki := ""
if isWiki {
wiki = "Wiki "
}
stderrBuilder.Reset()
stdoutBuilder.Reset()
pruneErr := git.NewCommand(ctx, "remote", "prune").AddDynamicArguments(m.GetRemoteName()).
SetDescription(fmt.Sprintf("Mirror.runSync %ssPrune references: %s ", wiki, m.Repo.FullName())).
Run(&git.RunOpts{
Timeout: timeout,
Dir: repoPath,
Stdout: stdoutBuilder,
Stderr: stderrBuilder,
})
if pruneErr != nil {
stdout := stdoutBuilder.String()
stderr := stderrBuilder.String()
// sanitize the output, since it may contain the remote address, which may
// contain a password
stderrMessage := util.SanitizeCredentialURLs(stderr)
stdoutMessage := util.SanitizeCredentialURLs(stdout)
log.Error("Failed to prune mirror repository %s%-v references:\nStdout: %s\nStderr: %s\nErr: %v", wiki, m.Repo, stdoutMessage, stderrMessage, pruneErr)
desc := fmt.Sprintf("Failed to prune mirror repository %s'%s' references: %s", wiki, repoPath, stderrMessage)
if err := system_model.CreateRepositoryNotice(desc); err != nil {
log.Error("CreateRepositoryNotice: %v", err)
}
// this if will only be reached on a successful prune so try to get the mirror again
}
return pruneErr
}
// runSync returns true if sync finished without error.
func runSync(ctx context.Context, m *repo_model.Mirror, delOldBranches bool) ([]*mirrorSyncResult, bool) {
repoPath := m.Repo.RepoPath()
wikiPath := m.Repo.WikiPath()
timeout := time.Duration(setting.Git.Timeout.Mirror) * time.Second
log.Trace("SyncMirrors [repo: %-v]: running git remote update...", m.Repo)
// use fetch but not remote update because git fetch support --tags but remote update doesn't
cmd := git.NewCommand(ctx, "fetch")
if m.EnablePrune {
cmd.AddArguments("--prune")
}
cmd.AddArguments("--tags").AddDynamicArguments(m.GetRemoteName())
remoteURL, remoteErr := git.GetRemoteURL(ctx, repoPath, m.GetRemoteName())
if remoteErr != nil {
log.Error("SyncMirrors [repo: %-v]: GetRemoteAddress Error %v", m.Repo, remoteErr)
return nil, false
}
envs := proxy.EnvWithProxy(remoteURL.URL)
stdoutBuilder := strings.Builder{}
stderrBuilder := strings.Builder{}
if err := cmd.
SetDescription(fmt.Sprintf("Mirror.runSync: %s", m.Repo.FullName())).
Run(&git.RunOpts{
Timeout: timeout,
Dir: repoPath,
Env: envs,
Stdout: &stdoutBuilder,
Stderr: &stderrBuilder,
}); err != nil {
stdout := stdoutBuilder.String()
stderr := stderrBuilder.String()
// sanitize the output, since it may contain the remote address, which may contain a password
stderrMessage := util.SanitizeCredentialURLs(stderr)
stdoutMessage := util.SanitizeCredentialURLs(stdout)
// Now check if the error is a resolve reference due to broken reference
if strings.Contains(stderr, "unable to resolve reference") && strings.Contains(stderr, "reference broken") {
log.Warn("SyncMirrors [repo: %-v]: failed to update mirror repository due to broken references:\nStdout: %s\nStderr: %s\nErr: %v\nAttempting Prune", m.Repo, stdoutMessage, stderrMessage, err)
err = nil
// Attempt prune
pruneErr := pruneBrokenReferences(ctx, m, repoPath, timeout, &stdoutBuilder, &stderrBuilder, false)
if pruneErr == nil {
// Successful prune - reattempt mirror
stderrBuilder.Reset()
stdoutBuilder.Reset()
if err = cmd.
SetDescription(fmt.Sprintf("Mirror.runSync: %s", m.Repo.FullName())).
Run(&git.RunOpts{
Timeout: timeout,
Dir: repoPath,
Stdout: &stdoutBuilder,
Stderr: &stderrBuilder,
}); err != nil {
stdout := stdoutBuilder.String()
stderr := stderrBuilder.String()
// sanitize the output, since it may contain the remote address, which may
// contain a password
stderrMessage = util.SanitizeCredentialURLs(stderr)
stdoutMessage = util.SanitizeCredentialURLs(stdout)
}
}
}
// If there is still an error (or there always was an error)
if err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to update mirror repository:\nStdout: %s\nStderr: %s\nErr: %v", m.Repo, stdoutMessage, stderrMessage, err)
desc := fmt.Sprintf("Failed to update mirror repository '%s': %s", repoPath, stderrMessage)
if err = system_model.CreateRepositoryNotice(desc); err != nil {
log.Error("CreateRepositoryNotice: %v", err)
}
return nil, false
}
}
output := stderrBuilder.String()
if err := git.WriteCommitGraph(ctx, repoPath); err != nil {
log.Error("SyncMirrors [repo: %-v]: %v", m.Repo, err)
}
gitRepo, err := git.OpenRepository(ctx, repoPath)
if err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to OpenRepository: %v", m.Repo, err)
return nil, false
}
log.Trace("SyncMirrors [repo: %-v]: syncing branches...", m.Repo)
if _, err = hat_repo_module.SyncRepoBranchesWithRepo(ctx, m.Repo, gitRepo, 0, delOldBranches); err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to synchronize branches: %v", m.Repo, err)
}
log.Trace("SyncMirrors [repo: %-v]: syncing releases with tags...", m.Repo)
if err = repo_module.SyncReleasesWithTags(ctx, m.Repo, gitRepo); err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to synchronize tags to releases: %v", m.Repo, err)
}
if m.LFS && setting.LFS.StartServer {
log.Trace("SyncMirrors [repo: %-v]: syncing LFS objects...", m.Repo)
endpoint := lfs.DetermineEndpoint(remoteURL.String(), m.LFSEndpoint)
lfsClient := lfs.NewClient(endpoint, nil)
if err = repo_module.StoreMissingLfsObjectsInRepository(ctx, m.Repo, gitRepo, lfsClient); err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to synchronize LFS objects for repository: %v", m.Repo, err)
}
}
gitRepo.Close()
log.Trace("SyncMirrors [repo: %-v]: updating size of repository", m.Repo)
if err := repo_module.UpdateRepoSize(ctx, m.Repo); err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to update size for mirror repository: %v", m.Repo, err)
}
if m.Repo.HasWiki() {
log.Trace("SyncMirrors [repo: %-v Wiki]: running git remote update...", m.Repo)
stderrBuilder.Reset()
stdoutBuilder.Reset()
if err := git.NewCommand(ctx, "remote", "update", "--prune").AddDynamicArguments(m.GetRemoteName()).
SetDescription(fmt.Sprintf("Mirror.runSync Wiki: %s ", m.Repo.FullName())).
Run(&git.RunOpts{
Timeout: timeout,
Dir: wikiPath,
Stdout: &stdoutBuilder,
Stderr: &stderrBuilder,
}); err != nil {
stdout := stdoutBuilder.String()
stderr := stderrBuilder.String()
// sanitize the output, since it may contain the remote address, which may contain a password
stderrMessage := util.SanitizeCredentialURLs(stderr)
stdoutMessage := util.SanitizeCredentialURLs(stdout)
// Now check if the error is a resolve reference due to broken reference
if strings.Contains(stderrMessage, "unable to resolve reference") && strings.Contains(stderrMessage, "reference broken") {
log.Warn("SyncMirrors [repo: %-v Wiki]: failed to update mirror wiki repository due to broken references:\nStdout: %s\nStderr: %s\nErr: %v\nAttempting Prune", m.Repo, stdoutMessage, stderrMessage, err)
err = nil
// Attempt prune
pruneErr := pruneBrokenReferences(ctx, m, repoPath, timeout, &stdoutBuilder, &stderrBuilder, true)
if pruneErr == nil {
// Successful prune - reattempt mirror
stderrBuilder.Reset()
stdoutBuilder.Reset()
if err = git.NewCommand(ctx, "remote", "update", "--prune").AddDynamicArguments(m.GetRemoteName()).
SetDescription(fmt.Sprintf("Mirror.runSync Wiki: %s ", m.Repo.FullName())).
Run(&git.RunOpts{
Timeout: timeout,
Dir: wikiPath,
Stdout: &stdoutBuilder,
Stderr: &stderrBuilder,
}); err != nil {
stdout := stdoutBuilder.String()
stderr := stderrBuilder.String()
stderrMessage = util.SanitizeCredentialURLs(stderr)
stdoutMessage = util.SanitizeCredentialURLs(stdout)
}
}
}
// If there is still an error (or there always was an error)
if err != nil {
log.Error("SyncMirrors [repo: %-v Wiki]: failed to update mirror repository wiki:\nStdout: %s\nStderr: %s\nErr: %v", m.Repo, stdoutMessage, stderrMessage, err)
desc := fmt.Sprintf("Failed to update mirror repository wiki '%s': %s", wikiPath, stderrMessage)
if err = system_model.CreateRepositoryNotice(desc); err != nil {
log.Error("CreateRepositoryNotice: %v", err)
}
return nil, false
}
if err := git.WriteCommitGraph(ctx, wikiPath); err != nil {
log.Error("SyncMirrors [repo: %-v]: %v", m.Repo, err)
}
}
log.Trace("SyncMirrors [repo: %-v Wiki]: git remote update complete", m.Repo)
}
log.Trace("SyncMirrors [repo: %-v]: invalidating mirror branch caches...", m.Repo)
branches, _, err := git.GetBranchesByPath(ctx, m.Repo.RepoPath(), 0, 0)
if err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to GetBranches: %v", m.Repo, err)
return nil, false
}
for _, branch := range branches {
cache.Remove(m.Repo.GetCommitsCountCacheKey(branch.Name, true))
}
m.UpdatedUnix = timeutil.TimeStampNow()
return parseRemoteUpdateOutput(output, m.GetRemoteName()), true
}
// SyncPullMirror starts the sync of the pull mirror and schedules the next run.
func SyncPullMirror(ctx context.Context, repoID int64, delOldBranches bool) bool {
log.Trace("SyncMirrors [repo_id: %v]", repoID)
defer func() {
err := recover()
if err == nil {
return
}
// There was a panic whilst syncMirrors...
log.Error("PANIC whilst SyncMirrors[repo_id: %d] Panic: %v\nStacktrace: %s", repoID, err, log.Stack(2))
}()
m, err := repo_model.GetMirrorByRepoID(ctx, repoID)
if err != nil {
log.Error("SyncMirrors [repo_id: %v]: unable to GetMirrorByRepoID: %v", repoID, err)
return false
}
_ = m.GetRepository(ctx) // force load repository of mirror
ctx, _, finished := process.GetManager().AddContext(ctx, fmt.Sprintf("Syncing Mirror %s/%s", m.Repo.OwnerName, m.Repo.Name))
defer finished()
log.Trace("SyncMirrors [repo: %-v]: Running Sync", m.Repo)
results, ok := runSync(ctx, m, delOldBranches)
if !ok {
if err = repo_model.TouchMirror(ctx, m); err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to TouchMirror: %v", m.Repo, err)
}
return false
}
log.Trace("SyncMirrors [repo: %-v]: Scheduling next update", m.Repo)
m.ScheduleNextUpdate()
if err = repo_model.UpdateMirror(ctx, m); err != nil {
log.Error("SyncMirrors [repo: %-v]: failed to UpdateMirror with next update date: %v", m.Repo, err)
return false
}
var gitRepo *git.Repository
if len(results) == 0 {
log.Trace("SyncMirrors [repo: %-v]: no branches updated", m.Repo)
} else {
log.Trace("SyncMirrors [repo: %-v]: %d branches updated", m.Repo, len(results))
gitRepo, err = git.OpenRepository(ctx, m.Repo.RepoPath())
if err != nil {
log.Error("SyncMirrors [repo: %-v]: unable to OpenRepository: %v", m.Repo, err)
return false
}
defer gitRepo.Close()
if ok := checkAndUpdateEmptyRepository(m, gitRepo, results); !ok {
return false
}
}
for _, result := range results {
// Discard GitHub pull requests, i.e. refs/pull/*
if result.refName.IsPull() {
continue
}
// Create reference
if result.oldCommitID == gitShortEmptySha {
commitID, err := gitRepo.GetRefCommitID(result.refName.String())
if err != nil {
log.Error("SyncMirrors [repo: %-v]: unable to GetRefCommitID [ref_name: %s]: %v", m.Repo, result.refName, err)
continue
}
notify_service.SyncPushCommits(ctx, m.Repo.MustOwner(ctx), m.Repo, &repo_module.PushUpdateOptions{
RefFullName: result.refName,
OldCommitID: git.EmptySHA,
NewCommitID: commitID,
}, repo_module.NewPushCommits())
notify_service.SyncCreateRef(ctx, m.Repo.MustOwner(ctx), m.Repo, result.refName, commitID)
continue
}
// Delete reference
if result.newCommitID == gitShortEmptySha {
notify_service.SyncDeleteRef(ctx, m.Repo.MustOwner(ctx), m.Repo, result.refName)
continue
}
// Push commits
oldCommitID, err := git.GetFullCommitID(gitRepo.Ctx, gitRepo.Path, result.oldCommitID)
if err != nil {
log.Error("SyncMirrors [repo: %-v]: unable to get GetFullCommitID[%s]: %v", m.Repo, result.oldCommitID, err)
continue
}
newCommitID, err := git.GetFullCommitID(gitRepo.Ctx, gitRepo.Path, result.newCommitID)
if err != nil {
log.Error("SyncMirrors [repo: %-v]: unable to get GetFullCommitID [%s]: %v", m.Repo, result.newCommitID, err)
continue
}
commits, err := gitRepo.CommitsBetweenIDs(newCommitID, oldCommitID)
if err != nil {
log.Error("SyncMirrors [repo: %-v]: unable to get CommitsBetweenIDs [new_commit_id: %s, old_commit_id: %s]: %v", m.Repo, newCommitID, oldCommitID, err)
continue
}
theCommits := repo_module.GitToPushCommits(commits)
if len(theCommits.Commits) > setting.UI.FeedMaxCommitNum {
theCommits.Commits = theCommits.Commits[:setting.UI.FeedMaxCommitNum]
}
if newCommit, err := gitRepo.GetCommit(newCommitID); err != nil {
log.Error("SyncMirrors [repo: %-v]: unable to get commit %s: %v", m.Repo, newCommitID, err)
continue
} else {
theCommits.HeadCommit = repo_module.CommitToPushCommit(newCommit)
}
theCommits.CompareURL = m.Repo.ComposeCompareURL(oldCommitID, newCommitID)
notify_service.SyncPushCommits(ctx, m.Repo.MustOwner(ctx), m.Repo, &repo_module.PushUpdateOptions{
RefFullName: result.refName,
OldCommitID: oldCommitID,
NewCommitID: newCommitID,
}, theCommits)
}
log.Trace("SyncMirrors [repo: %-v]: done notifying updated branches/tags - now updating last commit time", m.Repo)
// Get latest commit date and update to current repository updated time
commitDate, err := git.GetLatestCommitTime(ctx, m.Repo.RepoPath())
if err != nil {
log.Error("SyncMirrors [repo: %-v]: unable to GetLatestCommitDate: %v", m.Repo, err)
return false
}
if err = repo_model.UpdateRepositoryUpdatedTime(ctx, m.RepoID, commitDate); err != nil {
log.Error("SyncMirrors [repo: %-v]: unable to update repository 'updated_unix': %v", m.Repo, err)
return false
}
log.Trace("SyncMirrors [repo: %-v]: Successfully updated", m.Repo)
return true
}
func checkAndUpdateEmptyRepository(m *repo_model.Mirror, gitRepo *git.Repository, results []*mirrorSyncResult) bool {
if !m.Repo.IsEmpty {
return true
}
hasDefault := false
hasMaster := false
hasMain := false
defaultBranchName := m.Repo.DefaultBranch
if len(defaultBranchName) == 0 {
defaultBranchName = setting.Repository.DefaultBranch
}
firstName := ""
for _, result := range results {
if !result.refName.IsBranch() {
continue
}
name := result.refName.BranchName()
if len(firstName) == 0 {
firstName = name
}
hasDefault = hasDefault || name == defaultBranchName
hasMaster = hasMaster || name == "master"
hasMain = hasMain || name == "main"
}
if len(firstName) > 0 {
if hasDefault {
m.Repo.DefaultBranch = defaultBranchName
} else if hasMaster {
m.Repo.DefaultBranch = "master"
} else if hasMain {
m.Repo.DefaultBranch = "main"
} else {
m.Repo.DefaultBranch = firstName
}
// Update the git repository default branch
if err := gitRepo.SetDefaultBranch(m.Repo.DefaultBranch); err != nil {
if !git.IsErrUnsupportedVersion(err) {
log.Error("Failed to update default branch of underlying git repository %-v. Error: %v", m.Repo, err)
desc := fmt.Sprintf("Failed to uupdate default branch of underlying git repository '%s': %v", m.Repo.RepoPath(), err)
if err = system_model.CreateRepositoryNotice(desc); err != nil {
log.Error("CreateRepositoryNotice: %v", err)
}
return false
}
}
m.Repo.IsEmpty = false
// Update the is empty and default_branch columns
if err := repo_model.UpdateRepositoryCols(db.DefaultContext, m.Repo, "default_branch", "is_empty"); err != nil {
log.Error("Failed to update default branch of repository %-v. Error: %v", m.Repo, err)
desc := fmt.Sprintf("Failed to uupdate default branch of repository '%s': %v", m.Repo.RepoPath(), err)
if err = system_model.CreateRepositoryNotice(desc); err != nil {
log.Error("CreateRepositoryNotice: %v", err)
}
return false
}
}
return true
}

View File

@ -0,0 +1,267 @@
// Copyright 2021 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package mirror
import (
"context"
"errors"
"fmt"
"io"
"regexp"
"strings"
"time"
repo_model "code.gitea.io/gitea/models/repo"
"code.gitea.io/gitea/modules/git"
"code.gitea.io/gitea/modules/lfs"
"code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/process"
"code.gitea.io/gitea/modules/repository"
"code.gitea.io/gitea/modules/setting"
"code.gitea.io/gitea/modules/timeutil"
"code.gitea.io/gitea/modules/util"
)
var stripExitStatus = regexp.MustCompile(`exit status \d+ - `)
// AddPushMirrorRemote registers the push mirror remote.
func AddPushMirrorRemote(ctx context.Context, m *repo_model.PushMirror, addr string) error {
addRemoteAndConfig := func(addr, path string) error {
cmd := git.NewCommand(ctx, "remote", "add", "--mirror=push").AddDynamicArguments(m.RemoteName, addr)
if strings.Contains(addr, "://") && strings.Contains(addr, "@") {
cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=push %s [repo_path: %s]", m.RemoteName, util.SanitizeCredentialURLs(addr), path))
} else {
cmd.SetDescription(fmt.Sprintf("remote add %s --mirror=push %s [repo_path: %s]", m.RemoteName, addr, path))
}
if _, _, err := cmd.RunStdString(&git.RunOpts{Dir: path}); err != nil {
return err
}
if _, _, err := git.NewCommand(ctx, "config", "--add").AddDynamicArguments("remote."+m.RemoteName+".push", "+refs/heads/*:refs/heads/*").RunStdString(&git.RunOpts{Dir: path}); err != nil {
return err
}
if _, _, err := git.NewCommand(ctx, "config", "--add").AddDynamicArguments("remote."+m.RemoteName+".push", "+refs/tags/*:refs/tags/*").RunStdString(&git.RunOpts{Dir: path}); err != nil {
return err
}
return nil
}
if err := addRemoteAndConfig(addr, m.Repo.RepoPath()); err != nil {
return err
}
if m.Repo.HasWiki() {
wikiRemoteURL := repository.WikiRemoteURL(ctx, addr)
if len(wikiRemoteURL) > 0 {
if err := addRemoteAndConfig(wikiRemoteURL, m.Repo.WikiPath()); err != nil {
return err
}
}
}
return nil
}
// RemovePushMirrorRemote removes the push mirror remote.
func RemovePushMirrorRemote(ctx context.Context, m *repo_model.PushMirror) error {
cmd := git.NewCommand(ctx, "remote", "rm").AddDynamicArguments(m.RemoteName)
_ = m.GetRepository()
if _, _, err := cmd.RunStdString(&git.RunOpts{Dir: m.Repo.RepoPath()}); err != nil {
return err
}
if m.Repo.HasWiki() {
if _, _, err := cmd.RunStdString(&git.RunOpts{Dir: m.Repo.WikiPath()}); err != nil {
// The wiki remote may not exist
log.Warn("Wiki Remote[%d] could not be removed: %v", m.ID, err)
}
}
return nil
}
// SyncPushMirror starts the sync of the push mirror and schedules the next run.
func SyncPushMirror(ctx context.Context, mirrorID int64) bool {
log.Trace("SyncPushMirror [mirror: %d]", mirrorID)
defer func() {
err := recover()
if err == nil {
return
}
// There was a panic whilst syncPushMirror...
log.Error("PANIC whilst syncPushMirror[%d] Panic: %v\nStacktrace: %s", mirrorID, err, log.Stack(2))
}()
m, err := repo_model.GetPushMirror(ctx, repo_model.PushMirrorOptions{ID: mirrorID})
if err != nil {
log.Error("GetPushMirrorByID [%d]: %v", mirrorID, err)
return false
}
_ = m.GetRepository()
m.LastError = ""
ctx, _, finished := process.GetManager().AddContext(ctx, fmt.Sprintf("Syncing PushMirror %s/%s to %s", m.Repo.OwnerName, m.Repo.Name, m.RemoteName))
defer finished()
log.Trace("SyncPushMirror [mirror: %d][repo: %-v]: Running Sync", m.ID, m.Repo)
err = runPushSync(ctx, m)
if err != nil {
log.Error("SyncPushMirror [mirror: %d][repo: %-v]: %v", m.ID, m.Repo, err)
m.LastError = stripExitStatus.ReplaceAllLiteralString(err.Error(), "")
}
m.LastUpdateUnix = timeutil.TimeStampNow()
if err := repo_model.UpdatePushMirror(ctx, m); err != nil {
log.Error("UpdatePushMirror [%d]: %v", m.ID, err)
return false
}
log.Trace("SyncPushMirror [mirror: %d][repo: %-v]: Finished", m.ID, m.Repo)
return err == nil
}
func runPushSync(ctx context.Context, m *repo_model.PushMirror) error {
timeout := time.Duration(setting.Git.Timeout.Mirror) * time.Second
performPush := func(path string) error {
remoteURL, err := git.GetRemoteURL(ctx, path, m.RemoteName)
if err != nil {
log.Error("GetRemoteAddress(%s) Error %v", path, err)
return errors.New("Unexpected error")
}
if setting.LFS.StartServer {
log.Trace("SyncMirrors [repo: %-v]: syncing LFS objects...", m.Repo)
gitRepo, err := git.OpenRepository(ctx, path)
if err != nil {
log.Error("OpenRepository: %v", err)
return errors.New("Unexpected error")
}
defer gitRepo.Close()
endpoint := lfs.DetermineEndpoint(remoteURL.String(), "")
lfsClient := lfs.NewClient(endpoint, nil)
if err := pushAllLFSObjects(ctx, gitRepo, lfsClient); err != nil {
return util.SanitizeErrorCredentialURLs(err)
}
}
log.Trace("Pushing %s mirror[%d] remote %s", path, m.ID, m.RemoteName)
if err := git.Push(ctx, path, git.PushOptions{
Remote: m.RemoteName,
Force: true,
Mirror: true,
Timeout: timeout,
}); err != nil {
log.Error("Error pushing %s mirror[%d] remote %s: %v", path, m.ID, m.RemoteName, err)
return util.SanitizeErrorCredentialURLs(err)
}
return nil
}
err := performPush(m.Repo.RepoPath())
if err != nil {
return err
}
if m.Repo.HasWiki() {
wikiPath := m.Repo.WikiPath()
_, err := git.GetRemoteAddress(ctx, wikiPath, m.RemoteName)
if err == nil {
err := performPush(wikiPath)
if err != nil {
return err
}
} else {
log.Trace("Skipping wiki: No remote configured")
}
}
return nil
}
func pushAllLFSObjects(ctx context.Context, gitRepo *git.Repository, lfsClient lfs.Client) error {
contentStore := lfs.NewContentStore()
pointerChan := make(chan lfs.PointerBlob)
errChan := make(chan error, 1)
go lfs.SearchPointerBlobs(ctx, gitRepo, pointerChan, errChan)
uploadObjects := func(pointers []lfs.Pointer) error {
err := lfsClient.Upload(ctx, pointers, func(p lfs.Pointer, objectError error) (io.ReadCloser, error) {
if objectError != nil {
return nil, objectError
}
content, err := contentStore.Get(p)
if err != nil {
log.Error("Error reading LFS object %v: %v", p, err)
}
return content, err
})
if err != nil {
select {
case <-ctx.Done():
return nil
default:
}
}
return err
}
var batch []lfs.Pointer
for pointerBlob := range pointerChan {
exists, err := contentStore.Exists(pointerBlob.Pointer)
if err != nil {
log.Error("Error checking if LFS object %v exists: %v", pointerBlob.Pointer, err)
return err
}
if !exists {
log.Trace("Skipping missing LFS object %v", pointerBlob.Pointer)
continue
}
batch = append(batch, pointerBlob.Pointer)
if len(batch) >= lfsClient.BatchSize() {
if err := uploadObjects(batch); err != nil {
return err
}
batch = nil
}
}
if len(batch) > 0 {
if err := uploadObjects(batch); err != nil {
return err
}
}
err, has := <-errChan
if has {
log.Error("Error enumerating LFS objects for repository: %v", err)
return err
}
return nil
}
func syncPushMirrorWithSyncOnCommit(ctx context.Context, repoID int64) {
pushMirrors, err := repo_model.GetPushMirrorsSyncedOnCommit(ctx, repoID)
if err != nil {
log.Error("repo_model.GetPushMirrorsSyncedOnCommit failed: %v", err)
return
}
for _, mirror := range pushMirrors {
AddPushMirrorToQueue(mirror.ID)
}
}

View File

@ -0,0 +1,31 @@
// Copyright 2022 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package mirror
import (
"context"
repo_model "code.gitea.io/gitea/models/repo"
user_model "code.gitea.io/gitea/models/user"
"code.gitea.io/gitea/modules/repository"
notify_service "code.gitea.io/gitea/services/notify"
)
func init() {
notify_service.RegisterNotifier(&mirrorNotifier{})
}
type mirrorNotifier struct {
notify_service.NullNotifier
}
var _ notify_service.Notifier = &mirrorNotifier{}
func (m *mirrorNotifier) PushCommits(ctx context.Context, _ *user_model.User, repo *repo_model.Repository, _ *repository.PushUpdateOptions, _ *repository.PushCommits) {
syncPushMirrorWithSyncOnCommit(ctx, repo.ID)
}
func (m *mirrorNotifier) SyncPushCommits(ctx context.Context, _ *user_model.User, repo *repo_model.Repository, _ *repository.PushUpdateOptions, _ *repository.PushCommits) {
syncPushMirrorWithSyncOnCommit(ctx, repo.ID)
}

72
services/mirror/queue.go Normal file
View File

@ -0,0 +1,72 @@
// Copyright 2022 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package mirror
import (
"code.gitea.io/gitea/modules/graceful"
"code.gitea.io/gitea/modules/log"
"code.gitea.io/gitea/modules/queue"
"code.gitea.io/gitea/modules/setting"
)
var mirrorQueue *queue.WorkerPoolQueue[*SyncRequest]
// SyncType type of sync request
type SyncType int
const (
// PullMirrorType for pull mirrors
PullMirrorType SyncType = iota
// PushMirrorType for push mirrors
PushMirrorType
)
// SyncRequest for the mirror queue
type SyncRequest struct {
Type SyncType
ReferenceID int64 // RepoID for pull mirror, MirrorID for push mirror
DelOldBranches bool // RepoID for pull mirror, MirrorID for push mirror
}
// StartSyncMirrors starts a go routine to sync the mirrors
func StartSyncMirrors(queueHandle func(data ...*SyncRequest) []*SyncRequest) {
if !setting.Mirror.Enabled {
return
}
mirrorQueue = queue.CreateUniqueQueue(graceful.GetManager().ShutdownContext(), "mirror-new", queueHandle)
if mirrorQueue == nil {
log.Fatal("Unable to create mirror queue")
}
go graceful.GetManager().RunWithCancel(mirrorQueue)
}
// AddPullMirrorToQueue adds repoID to mirror queue
func AddPullMirrorToQueue(repoID int64, delOldBranches bool) {
addMirrorToQueue(PullMirrorType, repoID, delOldBranches)
}
// AddPushMirrorToQueue adds the push mirror to the queue
func AddPushMirrorToQueue(mirrorID int64) {
addMirrorToQueue(PushMirrorType, mirrorID, false)
}
func addMirrorToQueue(syncType SyncType, referenceID int64, delOldBranches bool) {
if !setting.Mirror.Enabled {
return
}
go func() {
if err := PushToQueue(syncType, referenceID, delOldBranches); err != nil {
log.Error("Unable to push sync request for to the queue for pull mirror repo[%d]. Error: %v", referenceID, err)
}
}()
}
// PushToQueue adds the sync request to the queue
func PushToQueue(mirrorType SyncType, referenceID int64, delOldBranches bool) error {
return mirrorQueue.Push(&SyncRequest{
Type: mirrorType,
ReferenceID: referenceID,
DelOldBranches: delOldBranches,
})
}