krz/gitbay

Push and pull repository mirroring !12

merged merged by cmc on 2026-08-24 21:18 UTC · krz/gitbay:mirroring into main

15 files changed, +626 −0

Layout: unified · split

cmd/gitbay/main.go +7
@@ -243,6 +243,13 @@ func repoCmd() *cobra.Command {
243 pass("list", "list deploy keys", passOpts{server: []string{"repo", "deploy-key", "list"}, needsRepo: true}), 243 pass("list", "list deploy keys", passOpts{server: []string{"repo", "deploy-key", "list"}, needsRepo: true}),
244 pass("remove", "remove a deploy key: <fingerprint>", passOpts{server: []string{"repo", "deploy-key", "remove"}, needsRepo: true}), 244 pass("remove", "remove a deploy key: <fingerprint>", passOpts{server: []string{"repo", "deploy-key", "remove"}, needsRepo: true}),
245 ), 245 ),
246 group("mirror", "sync with a foreign remote",
247 pass("add", "add a mirror: <https-url> --direction push|pull [--username <u>] [--token-stdin]",
248 passOpts{server: []string{"repo", "mirror", "add"}, needsRepo: true, stdinOK: true}),
249 pass("list", "list mirrors with sync status", passOpts{server: []string{"repo", "mirror", "list"}, needsRepo: true}),
250 pass("remove", "remove a mirror: <id>", passOpts{server: []string{"repo", "mirror", "remove"}, needsRepo: true}),
251 pass("sync", "schedule an immediate sync", passOpts{server: []string{"repo", "mirror", "sync"}, needsRepo: true}),
252 ),
246 group("topics", "free-form repository tags", 253 group("topics", "free-form repository tags",
247 pass("list", "list topics", passOpts{server: []string{"repo", "topics"}, needsRepo: true}), 254 pass("list", "list topics", passOpts{server: []string{"repo", "topics"}, needsRepo: true}),
248 pass("add", "add topics: <topic>...", passOpts{server: []string{"repo", "topics", "add"}, needsRepo: true}), 255 pass("add", "add topics: <topic>...", passOpts{server: []string{"repo", "topics", "add"}, needsRepo: true}),
cmd/gitbayd/main.go +2
@@ -21,6 +21,7 @@ import (
21 "gitbay.org/gitbay/internal/config" 21 "gitbay.org/gitbay/internal/config"
22 "gitbay.org/gitbay/internal/control" 22 "gitbay.org/gitbay/internal/control"
23 "gitbay.org/gitbay/internal/mail" 23 "gitbay.org/gitbay/internal/mail"
24 "gitbay.org/gitbay/internal/mirror"
24 "gitbay.org/gitbay/internal/notify" 25 "gitbay.org/gitbay/internal/notify"
25 "gitbay.org/gitbay/internal/gitd" 26 "gitbay.org/gitbay/internal/gitd"
26 "gitbay.org/gitbay/internal/hookd" 27 "gitbay.org/gitbay/internal/hookd"
@@ -137,6 +138,7 @@ func serveCmd() *cobra.Command {
137 if cfg.Mail.SMTPHost != "" { 138 if cfg.Mail.SMTPHost != "" {
138 go notify.New(st, cfg, retryBase).Run(whCtx) 139 go notify.New(st, cfg, retryBase).Run(whCtx)
139 } 140 }
141 go mirror.New(st, cfg).Run(whCtx)
140 142
141 errCh := make(chan error, 3) 143 errCh := make(chan error, 3)
142 if cfg.SSH.Mode == "embedded" { 144 if cfg.SSH.Mode == "embedded" {
docs/admin.org +5
@@ -103,6 +103,11 @@ contradiction; =--no-host-checks= skips port/path probes.
103 Serves only public repositories that additionally ran 103 Serves only public repositories that additionally ran
104 =repo settings git-daemon <repo> on=. 104 =repo settings git-daemon <repo> on=.
105 105
106** [mirrors]
107- =pull_interval_minutes= (15) — how often pull mirrors fetch their
108 upstream. Push mirrors sync shortly after each local ref update.
109 Mirror URLs pass the same SSRF rules as webhook targets.
110
106** [go_import] 111** [go_import]
107Vanity Go module paths, one per line: ="host/module" = "owner/repo"=. 112Vanity Go module paths, one per line: ="host/module" = "owner/repo"=.
108Requests with =?go-get=1= at or under the module path answer with the 113Requests with =?go-get=1= at or under the module path answer with the
docs/users.org +13
@@ -127,6 +127,19 @@ gitbay repo import you/mirror --from https://github.com/you/repo.git \
127gitbay repo import-issues you/mirror --from you/repo --token-stdin 127gitbay repo import-issues you/mirror --from you/repo --token-stdin
128#+end_src 128#+end_src
129 129
130Mirroring keeps a foreign remote in sync during a gradual migration
131(repo admin; https remotes; the token is stored server-side for the
132recurring sync and never echoed back):
133
134#+begin_src sh
135gitbay repo mirror add you/project https://github.com/you/project.git \
136 --direction push --token-stdin # propagate after every local push
137gitbay repo mirror add you/copy https://github.com/them/theirs.git \
138 --direction pull # follow upstream; local pushes refused
139gitbay repo mirror list # sync status and last error, per mirror
140gitbay repo mirror sync / remove <id>
141#+end_src
142
130* Organizations 143* Organizations
131 144
132Orgs share the owner namespace with users and own repositories at 145Orgs share the owner namespace with users and own repositories at
e2e/mirror_test.go added +161
@@ -0,0 +1,161 @@
1package e2e
2
3import (
4 "net/http/cgi"
5 "net/http/httptest"
6 "os"
7 "os/exec"
8 "path/filepath"
9 "strings"
10 "testing"
11 "time"
12)
13
14// gitHTTPRemote serves a bare repository over smart HTTP (push enabled),
15// standing in for GitHub in mirror tests.
16func gitHTTPRemote(t *testing.T) (url, bareDir string) {
17 t.Helper()
18 parent := t.TempDir()
19 bareDir = filepath.Join(parent, "remote.git")
20 for _, args := range [][]string{
21 {"init", "--bare", "--initial-branch=main", bareDir},
22 {"-C", bareDir, "config", "http.receivepack", "true"},
23 } {
24 if out, err := exec.Command("git", args...).CombinedOutput(); err != nil {
25 t.Fatalf("git %v: %v\n%s", args, err, out)
26 }
27 }
28 execPath, err := exec.Command("git", "--exec-path").Output()
29 if err != nil {
30 t.Fatal(err)
31 }
32 h := &cgi.Handler{
33 Path: filepath.Join(strings.TrimSpace(string(execPath)), "git-http-backend"),
34 Env: []string{"GIT_PROJECT_ROOT=" + parent, "GIT_HTTP_EXPORT_ALL=1"},
35 }
36 srv := httptest.NewServer(h)
37 t.Cleanup(srv.Close)
38 return srv.URL + "/remote.git", bareDir
39}
40
41func waitFor(t *testing.T, what string, cond func() bool) {
42 t.Helper()
43 deadline := time.Now().Add(15 * time.Second)
44 for time.Now().Before(deadline) {
45 if cond() {
46 return
47 }
48 time.Sleep(150 * time.Millisecond)
49 }
50 t.Fatalf("timed out waiting for %s", what)
51}
52
53func TestMirrors(t *testing.T) {
54 t.Setenv("GITBAY_MIRROR_TICK", "200ms")
55 inst := startInstanceWith(t, "[webhooks]\nallow_local = true\n")
56 aliceKey := inst.newKey(t, "alice")
57 bobKey := inst.newKey(t, "bob")
58 inst.admin(t, "admin", "user", "create", "alice", "--key", aliceKey+".pub")
59 inst.admin(t, "admin", "user", "create", "bob", "--key", bobKey+".pub")
60
61 // ---- push mirror: local pushes propagate to the remote.
62 remoteURL, remoteBare := gitHTTPRemote(t)
63 if _, errOut, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 {
64 t.Fatalf("repo create: %s", errOut)
65 }
66 if _, _, code := inst.ssh(t, bobKey, "", "repo", "mirror", "add", "alice/app", remoteURL, "--direction", "push"); code != 4 {
67 t.Fatal("non-admin added a mirror")
68 }
69 if _, errOut, code := inst.ssh(t, aliceKey, "", "repo", "mirror", "add", "alice/app", remoteURL, "--direction", "push"); code != 0 {
70 t.Fatalf("mirror add: %s", errOut)
71 }
72
73 work := t.TempDir()
74 env := inst.gitEnv(aliceKey)
75 mustGit(t, work, env, "clone", inst.sshURL("alice/app"), "w")
76 dir := filepath.Join(work, "w")
77 os.WriteFile(filepath.Join(dir, "a.txt"), []byte("a\n"), 0o644)
78 mustGit(t, dir, env, "checkout", "-q", "-b", "main")
79 mustGit(t, dir, env, "add", ".")
80 mustGit(t, dir, env, "commit", "-q", "-m", "base")
81 mustGit(t, dir, env, "push", "-q", "origin", "main")
82 head := strings.TrimSpace(mustGit(t, dir, env, "rev-parse", "HEAD"))
83
84 waitFor(t, "push mirror sync", func() bool {
85 out, _ := exec.Command("git", "-C", remoteBare, "rev-parse", "refs/heads/main").Output()
86 return strings.TrimSpace(string(out)) == head
87 })
88 out, _, _ := inst.ssh(t, aliceKey, "", "repo", "mirror", "list", "alice/app", "--json")
89 if !strings.Contains(out, `"last_sync":"`) || strings.Contains(out, "token") ||
90 strings.Contains(out, `"last_error":"`) {
91 t.Fatalf("mirror list after sync: %s", out)
92 }
93
94 // ---- pull mirror: local repo follows the remote and refuses pushes.
95 srcURL, srcBare := gitHTTPRemote(t)
96 seed := t.TempDir()
97 mustGit(t, seed, env, "clone", "-q", srcBare, "s")
98 sdir := filepath.Join(seed, "s")
99 os.WriteFile(filepath.Join(sdir, "up.txt"), []byte("upstream\n"), 0o644)
100 mustGit(t, sdir, env, "checkout", "-q", "-b", "main")
101 mustGit(t, sdir, env, "add", ".")
102 mustGit(t, sdir, env, "commit", "-q", "-m", "upstream commit")
103 mustGit(t, sdir, env, "push", "-q", "origin", "main")
104 upstreamHead := strings.TrimSpace(mustGit(t, sdir, env, "rev-parse", "HEAD"))
105
106 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/follow"); code != 0 {
107 t.Fatal("repo create failed")
108 }
109 if _, errOut, code := inst.ssh(t, aliceKey, "", "repo", "mirror", "add", "alice/follow", srcURL, "--direction", "pull"); code != 0 {
110 t.Fatalf("pull mirror add: %s", errOut)
111 }
112 waitFor(t, "pull mirror sync", func() bool {
113 out, _, _ := inst.ssh(t, aliceKey, "", "repo", "log", "alice/follow", "--json")
114 return strings.Contains(out, upstreamHead)
115 })
116 // Local pushes are refused while the pull mirror exists.
117 work2 := t.TempDir()
118 mustGit(t, work2, env, "clone", "-q", inst.sshURL("alice/follow"), "f")
119 fdir := filepath.Join(work2, "f")
120 os.WriteFile(filepath.Join(fdir, "no.txt"), []byte("n\n"), 0o644)
121 mustGit(t, fdir, env, "add", ".")
122 mustGit(t, fdir, env, "commit", "-q", "-m", "local change")
123 if out, code := gitRun(t, fdir, env, "push", "origin", "HEAD:main"); code == 0 || !strings.Contains(out, "pull mirror") {
124 t.Fatalf("push to pull mirror: exit %d\n%s", code, out)
125 }
126 // Removing the mirror restores pushes.
127 out, _, _ = inst.ssh(t, aliceKey, "", "repo", "mirror", "list", "alice/follow", "--json")
128 id := out[strings.Index(out, `"id":`)+5:]
129 id = id[:strings.IndexAny(id, ",}")]
130 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "mirror", "remove", "alice/follow", strings.TrimSpace(id)); code != 0 {
131 t.Fatal("mirror remove failed")
132 }
133 mustGit(t, fdir, env, "push", "-q", "origin", "HEAD:main")
134
135 // ---- failure visibility: a dead remote records an error.
136 deadURL := remoteURL + "-gone"
137 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "mirror", "add", "alice/follow", deadURL, "--direction", "push"); code != 0 {
138 t.Fatal("dead mirror add failed")
139 }
140 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "mirror", "sync", "alice/follow"); code != 0 {
141 t.Fatal("mirror sync failed")
142 }
143 waitFor(t, "failure recorded", func() bool {
144 out, _, _ := inst.ssh(t, aliceKey, "", "repo", "mirror", "list", "alice/follow", "--json")
145 return strings.Contains(out, `"last_error":"git push`)
146 })
147}
148
149func TestMirrorSSRFGuard(t *testing.T) {
150 inst := startInstance(t) // default posture: allow_local off
151 aliceKey := inst.newKey(t, "alice")
152 inst.admin(t, "admin", "user", "create", "alice", "--key", aliceKey+".pub")
153 if _, _, code := inst.ssh(t, aliceKey, "", "repo", "create", "alice/app"); code != 0 {
154 t.Fatal("repo create failed")
155 }
156 _, errOut, code := inst.ssh(t, aliceKey, "", "repo", "mirror", "add", "alice/app",
157 "http://127.0.0.1:9999/x.git", "--direction", "push")
158 if code != 2 || !strings.Contains(errOut, "SSRF") {
159 t.Fatalf("local mirror allowed: exit %d, %s", code, errOut)
160 }
161}
internal/config/config.go +6
@@ -23,6 +23,7 @@ type Config struct {
23 Webhooks Webhooks `toml:"webhooks"` 23 Webhooks Webhooks `toml:"webhooks"`
24 Limits Limits `toml:"limits"` 24 Limits Limits `toml:"limits"`
25 Mail Mail `toml:"mail"` 25 Mail Mail `toml:"mail"`
26 Mirrors Mirrors `toml:"mirrors"`
26 // GoImport maps vanity Go module paths to repositories, e.g. 27 // GoImport maps vanity Go module paths to repositories, e.g.
27 // "gitbay.org/gitbay" = "krz/gitbay". Requests carrying ?go-get=1 28 // "gitbay.org/gitbay" = "krz/gitbay". Requests carrying ?go-get=1
28 // under a mapped path get a go-import meta tag. 29 // under a mapped path get a go-import meta tag.
@@ -80,6 +81,10 @@ type Webhooks struct {
80 AllowLocal bool `toml:"allow_local"` 81 AllowLocal bool `toml:"allow_local"`
81} 82}
82 83
84type Mirrors struct {
85 PullIntervalMinutes int `toml:"pull_interval_minutes"`
86}
87
83type Limits struct { 88type Limits struct {
84 MaxPackBytes int64 `toml:"max_pack_bytes"` 89 MaxPackBytes int64 `toml:"max_pack_bytes"`
85 MaxBlobBytes int64 `toml:"max_blob_bytes"` 90 MaxBlobBytes int64 `toml:"max_blob_bytes"`
@@ -106,6 +111,7 @@ func Default() Config {
106 Mode: "closed", 111 Mode: "closed",
107 }, 112 },
108 GitDaemon: GitDaemon{Port: 9418}, 113 GitDaemon: GitDaemon{Port: 9418},
114 Mirrors: Mirrors{PullIntervalMinutes: 15},
109 Limits: Limits{ 115 Limits: Limits{
110 MaxPackBytes: 2 << 30, // 2 GiB 116 MaxPackBytes: 2 << 30, // 2 GiB
111 MaxBlobBytes: 100 << 20, 117 MaxBlobBytes: 100 << 20,
internal/control/mirrorcmd.go added +176
@@ -0,0 +1,176 @@
1package control
2
3import (
4 "bufio"
5 "errors"
6 "fmt"
7 "io"
8 "strconv"
9 "strings"
10
11 "gitbay.org/gitbay/internal/policy"
12 "gitbay.org/gitbay/internal/protocol"
13 "gitbay.org/gitbay/internal/store"
14 "gitbay.org/gitbay/internal/webhook"
15)
16
17func init() {
18 register(Command{Path: []string{"repo", "mirror", "add"},
19 Summary: "mirror to or from a remote: repo mirror add <owner/name> <https-url> --direction push|pull [--username <u>] [--token-stdin]",
20 ReadsStdin: true, SSHOnly: true, Run: runMirrorAdd})
21 register(Command{Path: []string{"repo", "mirror", "list"},
22 Summary: "list mirrors with sync status: repo mirror list <owner/name>", ReadOnly: true, Run: runMirrorList})
23 register(Command{Path: []string{"repo", "mirror", "remove"},
24 Summary: "remove a mirror: repo mirror remove <owner/name> <id>", Run: runMirrorRemove})
25 register(Command{Path: []string{"repo", "mirror", "sync"},
26 Summary: "schedule an immediate sync: repo mirror sync <owner/name>", Run: runMirrorSync})
27}
28
29func runMirrorAdd(c *Ctx, args []string) int {
30 var path, urlArg, direction, username string
31 tokenStdin := false
32 for i := 0; i < len(args); i++ {
33 switch args[i] {
34 case "--direction", "--username":
35 if i+1 >= len(args) {
36 return c.fail(protocol.ExitUsage, "%s requires a value", args[i])
37 }
38 if args[i] == "--direction" {
39 direction = args[i+1]
40 } else {
41 username = args[i+1]
42 }
43 i++
44 case "--token-stdin":
45 tokenStdin = true
46 default:
47 if path == "" {
48 path = args[i]
49 } else if urlArg == "" {
50 urlArg = args[i]
51 } else {
52 return c.fail(protocol.ExitUsage, "unexpected argument %q", args[i])
53 }
54 }
55 }
56 if path == "" || urlArg == "" || (direction != "push" && direction != "pull") {
57 return c.fail(protocol.ExitUsage, "usage: repo mirror add <owner/name> <https-url> --direction push|pull [--username <u>] [--token-stdin]")
58 }
59 // The worker's git process dials this URL from the server: same SSRF
60 // surface as a webhook target, same rules.
61 if err := webhook.ValidateURL(urlArg, c.Cfg.Webhooks.AllowLocal); err != nil {
62 return c.fail(protocol.ExitUsage, "%v", err)
63 }
64 repo, code := resolveRepo(c, path, policy.CanAdmin)
65 if code >= 0 {
66 return code
67 }
68 token := ""
69 if tokenStdin {
70 line, err := bufio.NewReader(io.LimitReader(c.Stdin, 4096)).ReadString('\n')
71 if err != nil && line == "" {
72 return c.fail(protocol.ExitUsage, "--token-stdin given but stdin held no token")
73 }
74 token = strings.TrimSpace(line)
75 }
76 id, err := c.Store.AddMirror(repo.ID, direction, urlArg, username, token)
77 if err != nil {
78 if errors.Is(err, store.ErrExists) {
79 return c.fail(protocol.ExitUsage, "that mirror already exists")
80 }
81 return c.fail(protocol.ExitFailure, "%v", err)
82 }
83 note := ""
84 if direction == "pull" {
85 note = "; local pushes are now refused — refs come from the upstream"
86 }
87 return c.emit(map[string]any{"id": id, "direction": direction, "url": urlArg}, func(w io.Writer) {
88 fmt.Fprintf(w, "mirror %d added (%s %s)%s\n", id, direction, urlArg, note)
89 })
90}
91
92func runMirrorList(c *Ctx, args []string) int {
93 if len(args) != 1 {
94 return c.fail(protocol.ExitUsage, "usage: repo mirror list <owner/name>")
95 }
96 repo, code := resolveRepo(c, args[0], policy.CanAdmin)
97 if code >= 0 {
98 return code
99 }
100 ms, err := c.Store.ListMirrors(repo.ID)
101 if err != nil {
102 return c.fail(protocol.ExitFailure, "%v", err)
103 }
104 type out struct {
105 ID int64 `json:"id"`
106 Direction string `json:"direction"`
107 URL string `json:"url"`
108 Username string `json:"username,omitempty"`
109 Pending bool `json:"pending"`
110 LastSync string `json:"last_sync,omitempty"`
111 LastError string `json:"last_error,omitempty"`
112 }
113 var ds []out
114 for _, m := range ms {
115 // The token never leaves the server, in any encoding.
116 ds = append(ds, out{m.ID, m.Direction, m.URL, m.Username, m.Dirty, m.LastSync, m.LastError})
117 }
118 return c.emit(ds, func(w io.Writer) {
119 for _, d := range ds {
120 status := "ok"
121 if d.Pending {
122 status = "pending"
123 }
124 if d.LastError != "" {
125 status = "error: " + d.LastError
126 }
127 fmt.Fprintf(w, "%d\t%s\t%s\tlast %s\t%s\n", d.ID, d.Direction, d.URL, orDash(d.LastSync), status)
128 }
129 })
130}
131
132func orDash(s string) string {
133 if s == "" {
134 return "-"
135 }
136 return s
137}
138
139func runMirrorRemove(c *Ctx, args []string) int {
140 if len(args) != 2 {
141 return c.fail(protocol.ExitUsage, "usage: repo mirror remove <owner/name> <id>")
142 }
143 repo, code := resolveRepo(c, args[0], policy.CanAdmin)
144 if code >= 0 {
145 return code
146 }
147 id, err := strconv.ParseInt(args[1], 10, 64)
148 if err != nil {
149 return c.fail(protocol.ExitUsage, "bad mirror id %q", args[1])
150 }
151 if err := c.Store.RemoveMirror(repo.ID, id); err != nil {
152 if errors.Is(err, store.ErrNotFound) {
153 return c.fail(protocol.ExitNotFound, "no mirror %d on %s", id, repo.Path())
154 }
155 return c.fail(protocol.ExitFailure, "%v", err)
156 }
157 return c.emit(map[string]any{"removed": id}, func(w io.Writer) {
158 fmt.Fprintf(w, "removed mirror %d\n", id)
159 })
160}
161
162func runMirrorSync(c *Ctx, args []string) int {
163 if len(args) != 1 {
164 return c.fail(protocol.ExitUsage, "usage: repo mirror sync <owner/name>")
165 }
166 repo, code := resolveRepo(c, args[0], policy.CanAdmin)
167 if code >= 0 {
168 return code
169 }
170 if err := c.Store.MarkMirrorsDirty(repo.ID, ""); err != nil {
171 return c.fail(protocol.ExitFailure, "%v", err)
172 }
173 return c.emit(map[string]string{"sync": "scheduled"}, func(w io.Writer) {
174 fmt.Fprintln(w, "sync scheduled; check repo mirror list for the outcome")
175 })
176}
internal/control/mr.go +1
@@ -822,6 +822,7 @@ func runMRMerge(c *Ctx, args []string) int {
822 if mr.TargetRef == repo.DefaultBranch { 822 if mr.TargetRef == repo.DefaultBranch {
823 ProcessCommitMessages(c.Store, dir, repo, c.User.ID, targetSHA, newSHA) 823 ProcessCommitMessages(c.Store, dir, repo, c.User.ID, targetSHA, newSHA)
824 } 824 }
825 c.Store.MarkMirrorsDirty(repo.ID, "push")
825 if parts, err := c.Store.MRParticipants(mr.ID); err == nil { 826 if parts, err := c.Store.MRParticipants(mr.ID); err == nil {
826 notifyUsers(c, parts, mrSubject(repo, mr.Number, mr.Title), 827 notifyUsers(c, parts, mrSubject(repo, mr.Number, mr.Title),
827 notifyBody(c, fmt.Sprintf("merged !%d into %s (%s)", mr.Number, mr.TargetRef, strategy), "", fmt.Sprintf("%s/mrs/%d", repo.Path(), mr.Number))) 828 notifyBody(c, fmt.Sprintf("merged !%d into %s (%s)", mr.Number, mr.TargetRef, strategy), "", fmt.Sprintf("%s/mrs/%d", repo.Path(), mr.Number)))
internal/hookd/hookd.go +2
@@ -183,6 +183,8 @@ func (s *Server) postReceive(req Request) {
183 dir := control.RepoDir(s.cfg.Server.Root, pushedRepo.OwnerName, pushedRepo.Name) 183 dir := control.RepoDir(s.cfg.Server.Root, pushedRepo.OwnerName, pushedRepo.Name)
184 control.ProcessCommitMessages(s.st, dir, pushedRepo, req.UserID, u.Old, u.New) 184 control.ProcessCommitMessages(s.st, dir, pushedRepo, req.UserID, u.Old, u.New)
185 } 185 }
186 // Any branch/tag update schedules the push mirrors.
187 s.st.MarkMirrorsDirty(req.RepoID, "push")
186 mrs, err := s.st.OpenMRsBySource(req.RepoID, branch) 188 mrs, err := s.st.OpenMRsBySource(req.RepoID, branch)
187 if err != nil { 189 if err != nil {
188 slog.Error("post-receive: listing MRs", "err", err) 190 slog.Error("post-receive: listing MRs", "err", err)
internal/httpd/accounts.go +1
@@ -332,5 +332,6 @@ func (s *Server) editSubmit(w http.ResponseWriter, r *http.Request, u store.User
332 fail(err.Error()) 332 fail(err.Error())
333 return 333 return
334 } 334 }
335 s.st.MarkMirrorsDirty(repo.ID, "push")
335 http.Redirect(w, r, fmt.Sprintf("/%s/blob/%s/%s", repo.Path(), ref, filePath), http.StatusSeeOther) 336 http.Redirect(w, r, fmt.Sprintf("/%s/blob/%s/%s", repo.Path(), ref, filePath), http.StatusSeeOther)
336} 337}
internal/mirror/mirror.go added +115
@@ -0,0 +1,115 @@
1// Package mirror synchronizes repositories with foreign remotes: push
2// mirrors propagate local refs outward after each receive, pull mirrors
3// keep a local copy fresh from an upstream. Sync runs in a background
4// worker, never in the push path; outcomes are recorded per mirror so
5// `repo mirror list` shows failure states like webhook deliveries do.
6package mirror
7
8import (
9 "context"
10 "fmt"
11 "log/slog"
12 "os"
13 "os/exec"
14 "path/filepath"
15 "time"
16
17 "gitbay.org/gitbay/internal/config"
18 "gitbay.org/gitbay/internal/control"
19 "gitbay.org/gitbay/internal/store"
20)
21
22const askpassScript = `#!/bin/sh
23case "$1" in
24 Username*) echo "${GITBAY_MIRROR_USER}" ;;
25 *) echo "${GITBAY_MIRROR_TOKEN}" ;;
26esac
27`
28
29type Worker struct {
30 St *store.Store
31 Cfg config.Config
32 Tick time.Duration
33}
34
35func New(st *store.Store, cfg config.Config) *Worker {
36 tick := 10 * time.Second
37 if v := os.Getenv("GITBAY_MIRROR_TICK"); v != "" {
38 if d, err := time.ParseDuration(v); err == nil {
39 tick = d
40 }
41 }
42 return &Worker{St: st, Cfg: cfg, Tick: tick}
43}
44
45func (w *Worker) Run(ctx context.Context) {
46 t := time.NewTicker(w.Tick)
47 defer t.Stop()
48 for {
49 select {
50 case <-ctx.Done():
51 return
52 case <-t.C:
53 w.sweep()
54 }
55 }
56}
57
58func (w *Worker) sweep() {
59 interval := w.Cfg.Mirrors.PullIntervalMinutes * 60
60 due, err := w.St.DueMirrors(interval)
61 if err != nil {
62 slog.Error("mirror: listing due", "err", err)
63 return
64 }
65 for _, m := range due {
66 if err := w.sync(m); err != nil {
67 slog.Warn("mirror sync failed", "mirror", m.ID, "url", m.URL, "err", err)
68 w.St.SetMirrorResult(m.ID, err.Error())
69 } else {
70 w.St.SetMirrorResult(m.ID, "")
71 }
72 }
73}
74
75func (w *Worker) sync(m store.Mirror) error {
76 repo, err := w.St.RepoByID(m.RepoID)
77 if err != nil {
78 return err
79 }
80 dir := control.RepoDir(w.Cfg.Server.Root, repo.OwnerName, repo.Name)
81
82 env := []string{"GIT_TERMINAL_PROMPT=0", "HOME=" + w.Cfg.Server.Root}
83 if m.Token != "" {
84 askpass := filepath.Join(w.Cfg.Server.Root, "mirror-askpass.sh")
85 if err := os.WriteFile(askpass, []byte(askpassScript), 0o700); err != nil {
86 return err
87 }
88 user := m.Username
89 if user == "" {
90 user = "x-access-token"
91 }
92 env = append(env,
93 "GIT_ASKPASS="+askpass,
94 "GITBAY_MIRROR_USER="+user,
95 "GITBAY_MIRROR_TOKEN="+m.Token)
96 }
97
98 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
99 defer cancel()
100 var args []string
101 if m.Direction == "push" {
102 // Branches and tags only: internal refs (merge-requests) stay home.
103 args = []string{"-C", dir, "push", "--prune", m.URL,
104 "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*"}
105 } else {
106 args = []string{"-C", dir, "fetch", "--prune", m.URL,
107 "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*"}
108 }
109 cmd := exec.CommandContext(ctx, "git", args...)
110 cmd.Env = env
111 if out, err := cmd.CombinedOutput(); err != nil {
112 return fmt.Errorf("git %s: %v: %.300s", m.Direction, err, out)
113 }
114 return nil
115}
internal/sshd/sshd.go +6
@@ -302,6 +302,12 @@ func runGit(cfg config.Config, st *store.Store, user store.User, scope string, a
302 fmt.Fprintf(stderr, "%s is archived and read-only\n", repo.Path()) 302 fmt.Fprintf(stderr, "%s is archived and read-only\n", repo.Path())
303 return protocol.ExitDenied 303 return protocol.ExitDenied
304 } 304 }
305 if write {
306 if mirrored, err := st.PullMirrored(repo.ID); err == nil && mirrored {
307 fmt.Fprintf(stderr, "%s is a pull mirror: its refs come from the upstream; push there instead\n", repo.Path())
308 return protocol.ExitDenied
309 }
310 }
305 311
306 dir := control.RepoDir(cfg.Server.Root, repo.OwnerName, repo.Name) 312 dir := control.RepoDir(cfg.Server.Root, repo.OwnerName, repo.Name)
307 env := []string{ 313 env := []string{
internal/store/migrations/0017_mirrors.down.sql added +1
@@ -0,0 +1 @@
1DROP TABLE mirrors;
internal/store/migrations/0017_mirrors.up.sql added +13
@@ -0,0 +1,13 @@
1CREATE TABLE mirrors (
2 id INTEGER PRIMARY KEY,
3 repo_id INTEGER NOT NULL REFERENCES repos(id) ON DELETE CASCADE,
4 direction TEXT NOT NULL CHECK (direction IN ('push','pull')),
5 url TEXT NOT NULL,
6 username TEXT NOT NULL DEFAULT '',
7 token TEXT NOT NULL DEFAULT '',
8 dirty INTEGER NOT NULL DEFAULT 1,
9 last_sync TEXT NOT NULL DEFAULT '',
10 last_error TEXT NOT NULL DEFAULT '',
11 created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
12 UNIQUE (repo_id, direction, url)
13);
internal/store/mirrors.go added +117
@@ -0,0 +1,117 @@
1package store
2
3import "errors"
4
5// ErrExists marks unique-constraint refusals callers turn into messages.
6var ErrExists = errors.New("already exists")
7
8// Mirror propagates refs to (push) or from (pull) a foreign remote. The
9// token is stored server-side — unlike import, mirroring is recurring —
10// and must never be echoed back in listings.
11type Mirror struct {
12 ID int64
13 RepoID int64
14 Direction string // push | pull
15 URL string
16 Username string
17 Token string
18 Dirty bool
19 LastSync string
20 LastError string
21}
22
23func (s *Store) AddMirror(repoID int64, direction, url, username, token string) (int64, error) {
24 res, err := s.DB.Exec(
25 "INSERT INTO mirrors (repo_id, direction, url, username, token) VALUES (?, ?, ?, ?, ?)",
26 repoID, direction, url, username, token)
27 if err != nil {
28 if isUniqueErr(err) {
29 return 0, ErrExists
30 }
31 return 0, err
32 }
33 return res.LastInsertId()
34}
35
36const mirrorSelect = `
37 SELECT id, repo_id, direction, url, username, token, dirty, last_sync, last_error
38 FROM mirrors`
39
40func scanMirror(row interface{ Scan(...any) error }) (Mirror, error) {
41 var m Mirror
42 err := row.Scan(&m.ID, &m.RepoID, &m.Direction, &m.URL, &m.Username, &m.Token,
43 &m.Dirty, &m.LastSync, &m.LastError)
44 return m, err
45}
46
47func (s *Store) mirrorQuery(q string, args ...any) ([]Mirror, error) {
48 rows, err := s.DB.Query(q, args...)
49 if err != nil {
50 return nil, err
51 }
52 defer rows.Close()
53 var out []Mirror
54 for rows.Next() {
55 m, err := scanMirror(rows)
56 if err != nil {
57 return nil, err
58 }
59 out = append(out, m)
60 }
61 return out, rows.Err()
62}
63
64func (s *Store) ListMirrors(repoID int64) ([]Mirror, error) {
65 return s.mirrorQuery(mirrorSelect+" WHERE repo_id = ? ORDER BY id", repoID)
66}
67
68// DueMirrors returns mirrors needing a sync: anything dirty, plus pull
69// mirrors whose last sync is older than intervalSeconds.
70func (s *Store) DueMirrors(intervalSeconds int) ([]Mirror, error) {
71 return s.mirrorQuery(mirrorSelect+`
72 WHERE dirty = 1
73 OR (direction = 'pull' AND (last_sync = ''
74 OR strftime('%s','now') - strftime('%s', last_sync) > ?))
75 ORDER BY id`, intervalSeconds)
76}
77
78func (s *Store) RemoveMirror(repoID, id int64) error {
79 res, err := s.DB.Exec("DELETE FROM mirrors WHERE repo_id = ? AND id = ?", repoID, id)
80 if err != nil {
81 return err
82 }
83 if n, _ := res.RowsAffected(); n == 0 {
84 return ErrNotFound
85 }
86 return nil
87}
88
89// MarkMirrorsDirty schedules a sync. An empty direction marks both.
90func (s *Store) MarkMirrorsDirty(repoID int64, direction string) error {
91 q := "UPDATE mirrors SET dirty = 1 WHERE repo_id = ?"
92 args := []any{repoID}
93 if direction != "" {
94 q += " AND direction = ?"
95 args = append(args, direction)
96 }
97 _, err := s.DB.Exec(q, args...)
98 return err
99}
100
101// SetMirrorResult records a sync outcome and clears the dirty flag.
102func (s *Store) SetMirrorResult(id int64, syncErr string) error {
103 _, err := s.DB.Exec(`
104 UPDATE mirrors SET dirty = 0, last_error = ?,
105 last_sync = strftime('%Y-%m-%dT%H:%M:%fZ','now')
106 WHERE id = ?`, syncErr, id)
107 return err
108}
109
110// PullMirrored reports whether the repo has a pull mirror, which makes it
111// read-only locally: its refs belong to the upstream.
112func (s *Store) PullMirrored(repoID int64) (bool, error) {
113 var n int
114 err := s.DB.QueryRow(
115 "SELECT COUNT(*) FROM mirrors WHERE repo_id = ? AND direction = 'pull'", repoID).Scan(&n)
116 return n > 0, err
117}