path: root/sync.go

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
}