From 3284b1c4906b90ca2e3864937c66d0fa30772ebf Mon Sep 17 00:00:00 2001 From: Lena Date: Sat, 1 Aug 2026 00:00:00 +0000 Subject: Enter git-forge-cloudflare Runs git-forge on Cloudflare Containers with R2 for durable storage. A Worker routes every request to one container through a Durable Object. Inside it, syncd mirrors the data directory to the bucket and restores it at boot, holding a write until it is durable. git-forge is neither modified nor vendored: the image builds it from its own repository at the commit named in forge.rev. --- sync.go | 1043 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1043 insertions(+) create mode 100644 sync.go (limited to 'sync.go') diff --git a/sync.go b/sync.go new file mode 100644 index 0000000..47a2136 --- /dev/null +++ b/sync.go @@ -0,0 +1,1043 @@ +package main + +// The mirror. Objects are content addressed and append only; refs are the +// mutable part, stored as state rather than files, because a restored loose +// ref would win over packed-refs and resurrect an old head. + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io/fs" + "log" + "os" + "os/exec" + "path/filepath" + "slices" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" +) + +const ( + manifestKey = "manifest.json" + // staleKey records the directories the manifest stopped listing, and when. + staleKey = "stale.json" + // pruneGrace is how long an unlisted repository stays in the bucket: the + // window to put back a manifest restored from a backup or written short. + pruneGrace = 7 * 24 * time.Hour + // pruneEvery is how often a running container looks again, so one that + // never idles still sees the grace out. + pruneEvery = 24 * time.Hour +) + +// SYNCD_LOOSE_LIMIT overrides it, for the test. +const defaultLooseLimit = 500 + +type Sync struct { + r2 *R2 + config string + repos string + staging string + + // One flush at a time, each rescanning, so a request covers its own + // writes. + flushMu sync.Mutex + + stateMu sync.Mutex + manifest map[string]bool // directory names under repos/, as "foo.git" + stored map[string]bool // object keys the bucket already holds + listed map[string]bool // prefixes read back from the bucket once + sent map[string][32]byte // digest per mutable key, written only when it moves + counts map[string]int // object files on disk per prefix, to notice a gc + gone map[string]map[string]bool // keys gone from disk at the last pass, per prefix + + looseLimit int + + hydrating sync.Map // repository path to *sync.Mutex + + // What the idle alarm asks for before it stops the container. + inflight atomic.Int64 +} + +func newSync(r2 *R2, dir string) *Sync { + limit := defaultLooseLimit + if n, err := strconv.Atoi(os.Getenv("SYNCD_LOOSE_LIMIT")); err == nil && n > 0 { + limit = n + } + return &Sync{ + r2: r2, + config: filepath.Join(dir, "config"), + repos: filepath.Join(dir, "repos"), + staging: filepath.Join(dir, ".hydrate"), + + manifest: map[string]bool{}, + stored: map[string]bool{}, + listed: map[string]bool{}, + sent: map[string][32]byte{}, + counts: map[string]int{}, + gone: map[string]map[string]bool{}, + + looseLimit: limit, + } +} + +// hydrated reports whether a directory has been filled in: hydrateOne moves +// objects in last. +func (s *Sync) hydrated(dir string) bool { + _, err := os.Stat(filepath.Join(s.repos, dir, "objects")) + return err == nil +} + +// restore brings back config/ in full and every repository as an empty +// directory, which is all git-forge's startup scan looks at. +func (s *Sync) restore() error { + // Debris from a hydration killed part way. + if err := os.RemoveAll(s.staging); err != nil { + return err + } + for _, d := range []string{s.config, s.repos, s.staging} { + if err := os.MkdirAll(d, 0700); err != nil { + return err + } + } + + data, found, err := s.r2.get(manifestKey) + if err != nil { + return err + } + if !found { + // A forge with no repositories never writes a manifest, and its config + // still has to come back. But an absent manifest also reads as every + // repository deleted, so one is believed only while the bucket holds + // no objects. + keys, err := s.r2.list("repos/") + if err != nil { + return err + } + // Objects, not keys: an empty REFS from a repository created in the + // window before its manifest is not worth refusing a boot over. + objects := 0 + for _, key := range keys { + if strings.Contains(key, "/objects/") { + objects++ + } + } + if objects > 0 { + return fmt.Errorf("no %s, but the bucket holds %d objects under repos/", manifestKey, objects) + } + log.Print("no manifest, so there are no repositories to restore") + return s.restoreConfig() + } + var dirs []string + if err := json.Unmarshal(data, &dirs); err != nil { + return fmt.Errorf("%s: %v", manifestKey, err) + } + for _, d := range dirs { + if !filepath.IsLocal(d) { + return fmt.Errorf("%s: %q is not a directory under repos/", manifestKey, d) + } + if err := os.MkdirAll(filepath.Join(s.repos, d), 0700); err != nil { + return err + } + s.manifest[d] = true + } + return s.restoreConfig() +} + +// restoreConfig is eager: git-forge reads users.json and repos.json before it +// serves anything. +func (s *Sync) restoreConfig() error { + keys, err := s.r2.list("config/") + if err != nil { + return err + } + if len(keys) == 0 { + return nil + } + gitdir := filepath.Join(s.config, ".git") + if err := gitInit(s.config, false); err != nil { + return err + } + want := make([]string, 0, len(keys)) + for _, key := range keys { + rest := strings.TrimPrefix(key, "config/") + // REFS is replayed below, once the objects it names are present. + if rest == "REFS" { + continue + } + // Skipped, not refused: a half written file is a key nothing reads. + if configTemp(rest) { + continue + } + if !configKey(rest) { + return fmt.Errorf("%s: refusing a key syncd did not write", key) + } + want = append(want, key) + } + err = s.fetchAll(want, func(key string) string { + rest := strings.TrimPrefix(key, "config/") + // git/ rather than .git/, so a listing is not full of dotted paths. + if after, ok := strings.CutPrefix(rest, "git/"); ok { + return filepath.Join(gitdir, after) + } + return filepath.Join(s.config, rest) + }) + if err != nil { + return err + } + // From the keys fetched, never the directory: a file only disk holds must + // not be remembered as stored. + for _, key := range want { + rest := strings.TrimPrefix(key, "config/") + if strings.HasPrefix(rest, "git/") { + continue // an object is immutable, so nothing compares it again + } + data, err := os.ReadFile(filepath.Join(s.config, rest)) + if err != nil { + return err + } + s.remember(key, data) + } + if err := s.fetchRefs(gitdir, "config/REFS"); err != nil { + // The log is only a log: one missing an object must not stop the + // forge. The next commit starts a fresh history. + log.Printf("config: audit log not replayed, its history restarts here: %v", err) + } + // Or every restored file looks untracked and the next change commits the + // whole tree. + if _, err := git(gitdir, s.config, nil, "reset", "--mixed", "--quiet"); err != nil { + log.Printf("config: index not reset, the next commit will be noisy: %v", err) + } + return nil +} + +// parallelDo runs fn over items and returns the first error. +func parallelDo(items []string, fn func(string) error) error { + const workers = 16 + if len(items) == 0 { + return nil + } + + jobs := make(chan string) + var wg sync.WaitGroup + var mu sync.Mutex + var failed error + + for range workers { + wg.Add(1) + go func() { + defer wg.Done() + // Drained even after a failure, or the sender deadlocks. + for item := range jobs { + mu.Lock() + stop := failed != nil + mu.Unlock() + if stop { + continue + } + err := fn(item) + mu.Lock() + if err != nil && failed == nil { + failed = err + } + mu.Unlock() + } + }() + } + for _, item := range items { + jobs <- item + } + close(jobs) + wg.Wait() + return failed +} + +// checkKeys refuses a key whose path would leave the directory it is written +// to: only syncd writes this bucket. +func checkKeys(keys []string, prefix string) error { + for _, key := range keys { + if !filepath.IsLocal(strings.TrimPrefix(key, prefix)) { + return fmt.Errorf("%s: refusing a key that points outside %s", key, prefix) + } + } + return nil +} + +// configKey is a key syncd wrote: config/{file} or config/git/objects/{path}. +// Inside the directory is not enough: .git/config and .git/hooks are inside +// it, and git runs both. +func configKey(rest string) bool { + if after, ok := strings.CutPrefix(rest, "git/objects/"); ok { + return filepath.IsLocal(after) + } + return filepath.IsLocal(rest) && !strings.ContainsRune(rest, '/') && !strings.HasPrefix(rest, ".") +} + +// configTemp matches the {name}.tmp{n} git-forge writes before renaming over a +// config file. +func configTemp(name string) bool { + i := strings.LastIndex(name, ".tmp") + if i < 0 { + return false + } + _, err := strconv.Atoi(name[i+len(".tmp"):]) + return err == nil +} + +// fetchAll downloads objects and records that the bucket holds them. +func (s *Sync) fetchAll(keys []string, dest func(key string) string) error { + err := parallelDo(keys, func(key string) error { + return s.r2.getFile(key, dest(key)) + }) + if err != nil { + return err + } + s.stateMu.Lock() + for _, key := range keys { + s.stored[key] = true + } + s.stateMu.Unlock() + return nil +} + +// hydrate fills in one repository and its siblings. Safe for one already +// there, and concurrently. +func (s *Sync) hydrate(base string) error { + value, _ := s.hydrating.LoadOrStore(base, &sync.Mutex{}) + mu := value.(*sync.Mutex) + mu.Lock() + defer mu.Unlock() + + for _, suffix := range []string{".git", ".issues.git", ".wiki.git"} { + dir := base + suffix + s.stateMu.Lock() + want := s.manifest[dir] + s.stateMu.Unlock() + if !want || s.hydrated(dir) { + continue + } + start := time.Now() + n, err := s.hydrateOne(dir) + if err != nil { + return fmt.Errorf("hydrate %s: %v", dir, err) + } + log.Printf("hydrated %s, %d objects in %s", dir, n, time.Since(start).Round(time.Millisecond)) + } + return nil +} + +// hydrateOne builds a repository beside the placeholder and moves it in, +// objects last, so a flush never sees an objects directory with refs not yet +// replayed. Staged outside repos/, which scanRepos claims. +func (s *Sync) hydrateOne(dir string) (int, error) { + tmp, err := os.MkdirTemp(s.staging, "") + if err != nil { + return 0, err + } + // The empty shell after a rename, or the whole failed attempt. + defer os.RemoveAll(tmp) + + path := filepath.Join(tmp, "repo") + if err := gitInit(path, true); err != nil { + return 0, err + } + prefix := "repos/" + dir + "/objects/" + keys, err := s.r2.list(prefix) + if err != nil { + return 0, err + } + if err := checkKeys(keys, prefix); err != nil { + return 0, err + } + err = s.fetchAll(keys, func(key string) string { + return filepath.Join(path, "objects", strings.TrimPrefix(key, prefix)) + }) + if err != nil { + return 0, err + } + // Refs after the objects they name, the order git pushes in. + if err := s.fetchRefs(path, "repos/"+dir+"/REFS"); err != nil { + return 0, err + } + + // The placeholder itself is never moved: a flush that saw it gone would + // publish a deletion. + dest := filepath.Join(s.repos, dir) + entries, err := os.ReadDir(path) + if err != nil { + return 0, err + } + for _, e := range entries { + if e.Name() == "objects" { + continue + } + // Rename refuses an existing directory, so debris from a failed + // attempt would wedge every retry. + target := filepath.Join(dest, e.Name()) + if err := os.RemoveAll(target); err != nil { + return 0, err + } + if err := os.Rename(filepath.Join(path, e.Name()), target); err != nil { + return 0, err + } + } + return len(keys), os.Rename(filepath.Join(path, "objects"), filepath.Join(dest, "objects")) +} + +// fetchRefs replays stored ref state in one exec. +func (s *Sync) fetchRefs(gitdir, key string) error { + data, found, err := s.r2.get(key) + if err != nil || !found { + return err + } + s.remember(key, data) + var cmds bytes.Buffer + for line := range strings.Lines(string(data)) { + id, ref, ok := strings.Cut(strings.TrimSpace(line), " ") + if !ok { + continue + } + fmt.Fprintf(&cmds, "create %s %s\n", ref, id) + } + if cmds.Len() == 0 { + return nil + } + _, err = git(gitdir, "", cmds.Bytes(), "update-ref", "--stdin") + return err +} + +// flush makes everything on disk durable, walking the whole datadir whatever +// provoked it. +func (s *Sync) flush() error { + s.flushMu.Lock() + defer s.flushMu.Unlock() + + uploaded := 0 + n, err := s.flushConfig() + if err != nil { + return err + } + uploaded += n + + dirs, err := scanRepos(s.repos) + if err != nil { + return err + } + for _, dir := range dirs { + // A placeholder would publish empty refs over the real ones. + if !s.hydrated(dir) { + continue + } + n, err := s.flushRepo(dir) + if err != nil { + // A repository deleted under this pass is not a failure; the + // manifest below publishes it. + if !s.hydrated(dir) { + continue + } + return err + } + uploaded += n + } + + if s.manifestMoved(dirs) { + // Last, so a crash leaves a manifest naming only repositories whose + // objects are already up. + data, err := json.Marshal(dirs) + if err != nil { + return err + } + if err := s.r2.put(manifestKey, data); err != nil { + return err + } + // After the put, or one refused write would be permanent. + s.setManifest(dirs) + log.Printf("manifest: %d directories", len(dirs)) + } + if uploaded > 0 { + log.Printf("flushed %d objects", uploaded) + } + return nil +} + +// flushRepo reads the refs before walking the objects and writes them after: +// read afterwards, they could name objects of a push this pass never saw. +func (s *Sync) flushRepo(dir string) (int, error) { + path := filepath.Join(s.repos, dir) + refs, err := readRefs(path) + if err != nil { + return 0, err + } + n, err := s.flushObjects(path, "repos/"+dir+"/objects/") + if err != nil { + return n, err + } + // Empty refs are a repository with no branches, or one being deleted under + // the read; once its objects are gone too, nothing is published. + if len(refs) == 0 && !s.hydrated(dir) { + return n, nil + } + _, err = s.putIfMoved("repos/"+dir+"/REFS", refs) + return n, err +} + +// flushConfig mirrors the config directory, working tree first: a missing +// audit log must not take the account table with it. +func (s *Sync) flushConfig() (int, error) { + // Rewritten in place, so content decides, not presence. + n := 0 + entries, err := os.ReadDir(s.config) + if err != nil { + return n, err + } + for _, e := range entries { + if e.IsDir() || strings.HasPrefix(e.Name(), ".") || configTemp(e.Name()) { + continue + } + data, err := os.ReadFile(filepath.Join(s.config, e.Name())) + if err != nil { + return n, err + } + wrote, err := s.putIfMoved("config/"+e.Name(), data) + if err != nil { + return n, err + } + if wrote { + n++ + } + } + + gitdir := filepath.Join(s.config, ".git") + if _, err := os.Stat(gitdir); err != nil { + return n, nil // git would not initialise; git-forge logs that and runs on + } + // Refs read before the walk, as in flushRepo, and written after the tree: + // here the tree is what a restore reads. + refs, err := readRefs(gitdir) + if err != nil { + return n, err + } + objects, err := s.flushObjects(gitdir, "config/git/objects/") + n += objects + if err != nil { + return n, err + } + _, err = s.putIfMoved("config/REFS", refs) + return n, err +} + +// flushObjects uploads every object the bucket lacks. A detached gc can delete +// a loose object under the walk; git finishes the pack first, so walking again +// finds it. +func (s *Sync) flushObjects(gitdir, prefix string) (int, error) { + uploaded := 0 + for range 2 { + n, vanished, err := s.uploadObjects(gitdir, prefix) + uploaded += n + if err != nil { + return uploaded, err + } + if !vanished { + return uploaded, nil + } + } + return uploaded, fmt.Errorf("%s: objects kept vanishing under the walk", prefix) +} + +// uploadObjects sends what the bucket lacks and reports whether anything +// vanished under it. Objects are never rewritten, so presence is a complete +// answer. +func (s *Sync) uploadObjects(gitdir, prefix string) (int, bool, error) { + if err := s.learn(prefix); err != nil { + return 0, false, err + } + root := filepath.Join(gitdir, "objects") + local, vanished := 0, false + var pending []string + err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { + if err != nil { + // gc removed a fanout directory it emptied; the caller walks again. + if path != root && os.IsNotExist(err) { + vanished = true + return fs.SkipDir + } + return err + } + if d.IsDir() { + return nil + } + rel, relErr := filepath.Rel(root, path) + if relErr != nil { + return relErr + } + if !mirrorable(rel, root) { + return nil + } + local++ + s.stateMu.Lock() + have := s.stored[prefix+filepath.ToSlash(rel)] + s.stateMu.Unlock() + if !have { + pending = append(pending, rel) + } + return nil + }) + // Only the root can be missing: a repository deleted under this pass. + if err != nil && !os.IsNotExist(err) { + return 0, false, err + } + + uploaded := 0 + err = parallelDo(pending, func(rel string) error { + key := prefix + filepath.ToSlash(rel) + if err := s.r2.putFile(key, filepath.Join(root, rel)); err != nil { + if !os.IsNotExist(err) { + return err + } + // gc took it; the caller walks again. + s.stateMu.Lock() + vanished = true + s.stateMu.Unlock() + return nil + } + s.stateMu.Lock() + s.stored[key] = true + uploaded++ + s.stateMu.Unlock() + return nil + }) + if err != nil { + return 0, false, err + } + + // A count below the last pass means a gc ran. Uploads are done, so the + // pack is up before anything is removed. + s.stateMu.Lock() + shrank := s.counts[prefix] > local + s.counts[prefix] = local + // What the last pass left waiting is this pass's to delete. + waiting := len(s.gone[prefix]) > 0 + s.stateMu.Unlock() + if shrank || waiting { + return uploaded, vanished, s.reconcile(gitdir, prefix) + } + return uploaded, vanished, nil +} + +// reconcile removes from the bucket what is no longer on disk, so only for a +// repository restored in full. A key goes on the second pass that finds it +// gone: the pack that replaced it reaches the bucket only after the walk that +// found it. +func (s *Sync) reconcile(gitdir, prefix string) error { + root := filepath.Join(gitdir, "objects") + keys, err := s.r2.list(prefix) + if err != nil { + return err + } + s.stateMu.Lock() + seen := s.gone[prefix] + s.stateMu.Unlock() + + waiting := map[string]bool{} + removed := 0 + for _, key := range keys { + rel := strings.TrimPrefix(key, prefix) + _, err := os.Stat(filepath.Join(root, filepath.FromSlash(rel))) + if err == nil { + continue + } + // This removes durable state, so a disk that cannot answer stops it. + if !os.IsNotExist(err) { + return err + } + if !seen[key] { + waiting[key] = true + continue + } + if err := s.r2.delete(key); err != nil { + return err + } + s.stateMu.Lock() + delete(s.stored, key) + s.stateMu.Unlock() + removed++ + } + // Recorded after the deletes, so a failed one is retried. + s.stateMu.Lock() + s.gone[prefix] = waiting + s.stateMu.Unlock() + if removed > 0 { + log.Printf("reconciled %s, removed %d objects gc had replaced", prefix, removed) + } + return nil +} + +// pruneDeleted removes what the bucket holds for a repository the manifest has +// not listed for pruneGrace. Deleting one removes its directories, so nothing +// walks its prefix again and reconcile never sees it. The delay is what makes +// a regressed manifest recoverable rather than a deletion at the next boot. +// +// The comparison is between the bucket and its own manifest, never a listing of +// the disk. Under flushMu because a new repository's objects go up before the +// manifest naming it. +func (s *Sync) pruneDeleted() error { + s.flushMu.Lock() + defer s.flushMu.Unlock() + + keys, err := s.r2.list("repos/") + if err != nil { + return err + } + stale := map[string][]string{} + for _, key := range keys { + dir, ok := repoDirOf(strings.TrimPrefix(key, "repos/")) + if !ok { + continue // not a key syncd wrote, so not one for this to decide + } + s.stateMu.Lock() + listed := s.manifest[dir] + s.stateMu.Unlock() + if !listed { + stale[dir] = append(stale[dir], key) + } + } + + since := map[string]time.Time{} + data, found, err := s.r2.get(staleKey) + if err != nil { + return err + } + if found { + if err := json.Unmarshal(data, &since); err != nil { + return fmt.Errorf("%s: %v", staleKey, err) + } + } + now := time.Now().UTC().Truncate(time.Second) + waiting := map[string]time.Time{} + for dir, keys := range stale { + first, ok := since[dir] + if !ok { + first = now + } + if now.Sub(first) < pruneGrace { + waiting[dir] = first + log.Printf("stale: %s, %d objects no manifest lists, removed after %s", + dir, len(keys), first.Add(pruneGrace).Format(time.DateOnly)) + continue + } + if err := parallelDo(keys, s.r2.delete); err != nil { + return err + } + s.stateMu.Lock() + for _, key := range keys { + delete(s.stored, key) + } + s.stateMu.Unlock() + log.Printf("pruned %s, %d objects the manifest stopped listing on %s", + dir, len(keys), first.Format(time.DateOnly)) + } + + if len(waiting) == 0 { + if found { + return s.r2.delete(staleKey) + } + return nil + } + data, err = json.Marshal(waiting) + if err != nil { + return err + } + return s.r2.put(staleKey, data) +} + +// repoDirOf names the directory a key under repos/ belongs to, by scanRepos's +// rule. +func repoDirOf(rest string) (string, bool) { + parts := strings.SplitN(rest, "/", 3) + if strings.HasSuffix(parts[0], ".git") { + return parts[0], true + } + if len(parts) == 3 && strings.HasSuffix(parts[1], ".git") { + return parts[0] + "/" + parts[1], true + } + return "", false +} + +// looseObjects counts what is in the two character fanout directories. +func looseObjects(gitdir string) int { + root := filepath.Join(gitdir, "objects") + fanout, err := os.ReadDir(root) + if err != nil { + return 0 + } + n := 0 + for _, d := range fanout { + if !d.IsDir() || len(d.Name()) != 2 { + continue + } + inner, err := os.ReadDir(filepath.Join(root, d.Name())) + if err != nil { + continue + } + n += len(inner) + } + return n +} + +// compactConfig packs the audit log: git's own auto gc waits for thousands of +// loose objects, which is a slow boot. Repositories are git-forge's to gc. +func (s *Sync) compactConfig() { + gitdir := filepath.Join(s.config, ".git") + n := looseObjects(gitdir) + if n < s.looseLimit { + return + } + if _, err := git(gitdir, s.config, nil, "gc", "--quiet"); err != nil { + log.Printf("config: not packed, the audit log will keep growing: %v", err) + return + } + log.Printf("config: packed %d loose objects", n) +} + +func mirrorable(rel, root string) bool { + // Derived and rewritten in place by gc: the one thing here not immutable. + if strings.HasPrefix(rel, "info/") { + return false + } + // Every component: receive-pack's quarantine directory holds ordinarily + // named files. + for _, part := range strings.Split(filepath.ToSlash(rel), "/") { + if strings.HasPrefix(part, "tmp_") || strings.HasPrefix(part, ".") { + return false + } + } + base := filepath.Base(rel) + // receive-pack's .keep is transient; mirrored, it would stop gc repacking + // that pack for ever. + if strings.HasSuffix(base, ".keep") { + return false + } + // A pack is complete only once its index is beside it. + if strings.HasSuffix(base, ".pack") { + if _, err := os.Stat(filepath.Join(root, strings.TrimSuffix(rel, ".pack")+".idx")); err != nil { + return false + } + } + return true +} + +// learn reads what the bucket holds under a prefix, once per boot: after that +// syncd is the only writer. +func (s *Sync) learn(prefix string) error { + s.stateMu.Lock() + known := s.listed[prefix] + s.stateMu.Unlock() + if known { + return nil + } + keys, err := s.r2.list(prefix) + if err != nil { + return err + } + s.stateMu.Lock() + for _, k := range keys { + s.stored[k] = true + } + s.listed[prefix] = true + s.stateMu.Unlock() + return nil +} + +// readRefs is show-ref's output. Exit 1 with nothing printed is no refs; +// everything else is an error, since this is written straight to the bucket. +func readRefs(gitdir string) ([]byte, error) { + out, err := git(gitdir, "", nil, "show-ref") + if err == nil { + return out, nil + } + var exit *exec.ExitError + if errors.As(err, &exit) && exit.ExitCode() == 1 && len(bytes.TrimSpace(out)) == 0 { + return nil, nil + } + return nil, fmt.Errorf("%s: %v", gitdir, err) +} + +// remember records what the bucket holds for a key, so the flush after a +// restore does not rewrite it. +func (s *Sync) remember(key string, data []byte) { + sum := sha256.Sum256(data) + s.stateMu.Lock() + s.sent[key] = sum + s.stateMu.Unlock() +} + +// putIfMoved writes a mutable key only when its content changed. +func (s *Sync) putIfMoved(key string, data []byte) (bool, error) { + sum := sha256.Sum256(data) + s.stateMu.Lock() + seen, ok := s.sent[key] + s.stateMu.Unlock() + if ok && seen == sum { + return false, nil + } + if err := s.r2.put(key, data); err != nil { + return false, err + } + s.stateMu.Lock() + s.sent[key] = sum + s.stateMu.Unlock() + return true, nil +} + +// manifestMoved reports whether the directory set differs from the bucket's. +func (s *Sync) manifestMoved(dirs []string) bool { + s.stateMu.Lock() + defer s.stateMu.Unlock() + if len(dirs) != len(s.manifest) { + return true + } + for _, d := range dirs { + if !s.manifest[d] { + return true + } + } + return false +} + +func (s *Sync) setManifest(dirs []string) { + next := make(map[string]bool, len(dirs)) + for _, d := range dirs { + next[d] = true + } + s.stateMu.Lock() + s.manifest = next + s.stateMu.Unlock() +} + +func (s *Sync) known(base string) bool { + s.stateMu.Lock() + defer s.stateMu.Unlock() + return s.manifest[base+".git"] +} + +// scanRepos lists every git directory under repos/ by git-forge's rule, +// siblings included: they have to be restored too. +func scanRepos(root string) ([]string, error) { + // Never nil: null in the manifest would read as no answer. + dirs := []string{} + top, err := os.ReadDir(root) + if err != nil { + return nil, err + } + for _, entry := range top { + if !entry.IsDir() { + continue + } + if strings.HasSuffix(entry.Name(), ".git") { + dirs = append(dirs, entry.Name()) + continue + } + inner, err := os.ReadDir(filepath.Join(root, entry.Name())) + if err != nil { + // A section can go between the two listings. + if os.IsNotExist(err) { + continue + } + return nil, err + } + for _, e := range inner { + if e.IsDir() && strings.HasSuffix(e.Name(), ".git") { + dirs = append(dirs, entry.Name()+"/"+e.Name()) + } + } + } + slices.Sort(dirs) + return dirs, nil +} + +// sweep backstops the writes no request holds for: token telemetry, and +// anything proxy.go's rule misses. +func (s *Sync) sweep(ctx context.Context) { + // Not in restore: a bucket this cannot list still serves. + if err := s.pruneDeleted(); err != nil { + log.Printf("prune: %v", err) + } + pruned := time.Now() + + t := time.NewTicker(sweepInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + // Packing first, so the flush uploads the pack before removing + // what it replaced. + s.compactConfig() + if err := s.flush(); err != nil { + log.Printf("sweep: %v", err) + } + if time.Since(pruned) >= pruneEvery { + if err := s.pruneDeleted(); err != nil { + log.Printf("prune: %v", err) + } + pruned = time.Now() + } + } + } +} + +// git returns stdout alone; stderr goes into the error, since readRefs stores +// stdout straight into the bucket. System and global configuration are +// neutralised. +func git(gitdir, worktree string, stdin []byte, args ...string) ([]byte, error) { + cmd := exec.Command("git", args...) + cmd.Env = append(os.Environ(), + "GIT_CONFIG_NOSYSTEM=1", + "GIT_CONFIG_GLOBAL=/dev/null", + "GIT_DIR="+gitdir, + ) + if worktree != "" { + cmd.Env = append(cmd.Env, "GIT_WORK_TREE="+worktree) + } + if stdin != nil { + cmd.Stdin = bytes.NewReader(stdin) + } + var out, fail bytes.Buffer + cmd.Stdout, cmd.Stderr = &out, &fail + if err := cmd.Run(); err != nil { + // %w, so readRefs can ask git's exit status. + if msg := bytes.TrimSpace(fail.Bytes()); len(msg) > 0 { + return out.Bytes(), fmt.Errorf("git %s: %w: %s", strings.Join(args, " "), err, msg) + } + return out.Bytes(), fmt.Errorf("git %s: %w", strings.Join(args, " "), err) + } + return out.Bytes(), nil +} + +// gitInit matches git-forge: an empty template and an explicit branch. +func gitInit(dir string, bare bool) error { + tmpl, err := os.MkdirTemp("", "syncd-template") + if err != nil { + return err + } + defer os.RemoveAll(tmpl) + + args := []string{"init", "--quiet", "--template=" + tmpl, "--initial-branch=master"} + if bare { + args = append(args, "--bare") + } + cmd := exec.Command("git", append(args, dir)...) + cmd.Env = append(os.Environ(), "GIT_CONFIG_NOSYSTEM=1", "GIT_CONFIG_GLOBAL=/dev/null") + out, err := cmd.CombinedOutput() + if err != nil { + return fmt.Errorf("git init %s: %v: %s", dir, err, out) + } + return nil +} -- cgit v1.2.3