Enter git-forge-cloudflareHEADmaster

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.

1 files changed, 1043 insertions, 0 deletions

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
+}