package main
import (
"bytes"
"context"
"errors"
"fmt"
"log"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/zitcli"
)
// A repository that follows a Git upstream (#358). The forge keeps no
// Git repository of its own: every pull fetches into a fresh, shallow
// Git repository with nothing checked out, `zit import-git --into` adds what the store
// does not hold, and the directory is deleted. What the store already
// holds is zit's to know (its import map beside the store), so a fetch
// only needs the history since the last pull. When it is not enough zit
// says so (retry "deepen") and the pull fetches deeper, up to the whole
// history. Git mirrors never accept local writes.
// gitFetchTimeout bounds all the fetching one pull does, and
// gitImportTimeout one import into a store: a stalled one must not hold
// the store's slot or lease. gitFirstImportTimeout bounds the first
// import, which writes the whole history: the Linux kernel's 1.3
// million commits take about nine hours with sub-manifests.
const (
gitFetchTimeout = 30 * time.Minute
gitImportTimeout = 60 * time.Minute
gitFirstImportTimeout = 16 * time.Hour
)
// gitFlatLimit is the most manifest lines (commits times files at the
// default branch's tip) a first import writes as flat manifests. Every
// flat check-in lists the whole tree, so the kernel's would be some
// 10^11 lines, about seventy hours; past the limit the import uses
// sub-manifests whatever the form asked for. At the kernel's measured
// rate the limit is about half an hour of flat import.
const gitFlatLimit = 1_000_000_000
// gitFirstDepth is how many commits of each branch a pull fetches
// first, and gitMaxDepth the deepest it asks for before fetching the
// whole history. Most pulls find fewer new commits than gitFirstDepth
// and fetch once.
const (
gitFirstDepth = 64
gitMaxDepth = 4096
)
// gitFreeFloor is the free space a fetch needs before it starts: below
// it, a fetch could fill the disk the stores and JetStream live on.
const gitFreeFloor = 3 << 30
// gitEnv is the environment every git the forge runs gets: no global
// or system configuration (no hook, helper or url rewrite in the forge
// account's own setup reaches an upstream), no prompting, and only
// https, so an address cannot make git speak another protocol.
func gitEnv(protocols string) []string {
return []string{
"GIT_CONFIG_GLOBAL=/dev/null", "GIT_CONFIG_NOSYSTEM=1", "GIT_TERMINAL_PROMPT=0",
"GIT_CONFIG_COUNT=2",
"GIT_CONFIG_KEY_0=protocol.allow", "GIT_CONFIG_VALUE_0=never",
"GIT_CONFIG_KEY_1=protocol." + protocols + ".allow", "GIT_CONFIG_VALUE_1=always",
}
}
// gitWorkspace makes an empty directory for one fetch, under a
// directory of this process's own, and the func that removes it. The
// first call also removes what earlier processes left: a pull a
// restart cut short leaves its fetch behind, and only this process's
// pulls are running.
func (rg *registrar) gitWorkspace(name string) (string, func(), error) {
root := filepath.Join(rg.dir, "gitimport")
own := filepath.Join(root, strconv.Itoa(os.Getpid()))
rg.gitSweep.Do(func() {
entries, _ := os.ReadDir(root)
for _, e := range entries {
if e.Name() != filepath.Base(own) {
if err := os.RemoveAll(filepath.Join(root, e.Name())); err != nil {
log.Printf("git workspaces: %v", err)
}
}
}
})
if err := os.MkdirAll(own, 0o700); err != nil {
return "", nil, err
}
if free, ok := freeBytes(own); ok && free < gitFreeFloor {
return "", nil, fmt.Errorf("only %d MB free; a Git fetch needs %d MB", free>>20, gitFreeFloor>>20)
}
dir, err := os.MkdirTemp(own, name+".")
if err != nil {
return "", nil, err
}
return dir, func() {
if err := os.RemoveAll(dir); err != nil {
log.Printf("git workspace %s: %v", name, err)
}
}, nil
}
// gitProtocol is the one protocol git may use: https, or what a test
// serves its upstream over.
func (rg *registrar) gitProtocol() string {
if rg.gitTestProtocol != "" {
return rg.gitTestProtocol
}
return "https"
}
// gitFetch brings the upstream's branches and tags into the Git
// repository at dir, making it the first time, and points HEAD at the
// upstream's default branch so the import names it trunk. depth is the
// commits of history per branch to fetch; 0 is all of it, deepening a
// shallow repository to the whole history.
func (rg *registrar) gitFetch(ctx context.Context, dir, upstream string, depth int) error {
if rg.gitOnFetch != nil {
rg.gitOnFetch(depth)
}
env := gitEnv(rg.gitProtocol())
// A repository with a work tree, though nothing is checked out into
// it: zit's first import reads only from one (its --into also
// accepts a bare repository, but one shape serves both).
if _, err := os.Stat(filepath.Join(dir, ".git")); errors.Is(err, os.ErrNotExist) {
if _, err := runGit(ctx, env, "", "init", "-q", dir); err != nil {
return err
}
}
args := []string{"fetch", "-q", "--prune", "--no-tags", "--update-head-ok"}
if depth > 0 {
args = append(args, "--depth="+strconv.Itoa(depth))
} else if _, err := os.Stat(filepath.Join(dir, ".git", "shallow")); err == nil {
args = append(args, "--unshallow")
}
args = append(args, "--", upstream, "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*")
if _, err := runGit(ctx, env, dir, args...); err != nil {
return err
}
head := ""
if out, err := runGit(ctx, env, dir, "ls-remote", "--symref", "--", upstream, "HEAD"); err == nil {
for _, l := range strings.Split(out, "\n") {
if rest, ok := strings.CutPrefix(l, "ref: "); ok {
head, _, _ = strings.Cut(rest, "\t")
break
}
}
}
if head == "" {
// An upstream that does not say: the conventional names, in
// the order hosts default to them.
for _, b := range []string{"refs/heads/main", "refs/heads/master", "refs/heads/trunk"} {
if _, err := runGit(ctx, env, dir, "rev-parse", "-q", "--verify", b); err == nil {
head = b
break
}
}
}
if head == "" {
return errors.New("the upstream has no default branch to follow")
}
// The name is the upstream's to choose; only a branch is followed,
// and nothing that could read as an option reaches git.
if !strings.HasPrefix(head, "refs/heads/") || strings.ContainsAny(head, " \t\n") {
return fmt.Errorf("the upstream's default branch %q is not a branch", head)
}
_, err := runGit(ctx, env, dir, "symbolic-ref", "HEAD", head)
return err
}
// runGit runs git with env added, in its own process group so a
// deadline ends every child it started, and answers its output.
func runGit(ctx context.Context, env []string, dir string, args ...string) (string, error) {
cmd := exec.CommandContext(ctx, "git", args...)
cmd.Dir = dir
cmd.Env = append(withoutGitEnv(os.Environ()), env...)
zitcli.OwnGroup(cmd)
cmd.WaitDelay = 2 * time.Second
var out bytes.Buffer
cmd.Stdout, cmd.Stderr = &out, &out
if err := cmd.Run(); err != nil {
if ctx.Err() != nil {
return "", fmt.Errorf("git %s did not finish within %v", args[0], gitFetchTimeout)
}
return "", fmt.Errorf("git %s: %s", args[0], gitLast(out.String(), err))
}
return out.String(), nil
}
// withoutGitEnv drops the GIT_* variables the forge process inherited:
// one such as GIT_ALLOW_PROTOCOL or GIT_SSH_COMMAND would override what
// gitEnv sets.
func withoutGitEnv(env []string) []string {
out := env[:0:0]
for _, kv := range env {
if !strings.HasPrefix(kv, "GIT_") {
out = append(out, kv)
}
}
return out
}
// forgetGitDirs removes what the forge keeps on disk for a
// repository's Git mirror out: the Git repository it exports into,
// which is rebuilt from the store, so a purge takes it with it. A
// mirror in keeps nothing beyond the pull in flight.
func (rg *registrar) forgetGitDirs(name string) {
if rg.dir == "" || name == "" {
return
}
defer rg.gitLock(name)()
if err := os.RemoveAll(filepath.Join(rg.dir, "gitmirror", "exports", name)); err != nil {
log.Printf("git dirs %s: %v", name, err)
}
}
// gitLock is the per-repository lock a Git mirror's pull runs under: a
// manual pull and a scheduled one import one at a time rather than
// meeting zit's own lock and failing.
func (rg *registrar) gitLock(name string) func() {
v, _ := rg.gitLocks.LoadOrStore(name, &sync.Mutex{})
mu := v.(*sync.Mutex)
mu.Lock()
return mu.Unlock
}
// gitLast is git's last line of output, or the process error.
func gitLast(out string, err error) string {
lines := strings.Split(strings.TrimSpace(out), "\n")
for i := len(lines) - 1; i >= 0; i-- {
if l := strings.TrimSpace(lines[i]); l != "" && !strings.HasPrefix(l, "hint:") {
return l
}
}
return err.Error()
}
// importGitClone makes the store at spec.Path from the Git repository
// at spec.Clone: a fetch of the whole history, then a fresh import. The
// fetch is deleted either way; a failure also removes what was written.
func (rg *registrar) importGitClone(ctx context.Context, spec api.RepoSpec) error {
if err := os.MkdirAll(filepath.Dir(spec.Path), 0o700); err != nil {
return err
}
dir, done, err := rg.gitWorkspace(spec.Name)
if err != nil {
return err
}
defer done()
fctx, cancel := context.WithTimeout(ctx, gitFetchTimeout)
defer cancel()
if err := rg.gitFetch(fctx, dir, spec.Clone, 0); err != nil {
return err
}
// zit writes a fresh destination only; the empty directory a caller
// may have made for it (Add accepts one) goes first. Remove refuses
// a directory that is not empty, which Add already ruled out.
_ = os.Remove(spec.Path)
zitOnly := spec.ZitOnly
if !zitOnly {
limit := rg.gitFlatLimit
if limit == 0 {
limit = gitFlatLimit
}
if lines := gitManifestLines(ctx, gitEnv(rg.gitProtocol()), dir); lines > limit {
log.Printf("import %s: %d manifest lines flat, over %d: importing with sub-manifests", spec.Name, lines, limit)
zitOnly = true
}
}
ictx, icancel := context.WithTimeout(ctx, gitFirstImportTimeout)
defer icancel()
r, err := zitcli.ImportGit(ictx, dir, spec.Path, gitEnv(rg.gitProtocol()), zitOnly)
if err == nil {
err = r.Err()
}
if err != nil {
return err
}
log.Printf("import %s from Git %s: %d check-ins", spec.Name, spec.Clone, r.Added)
return nil
}
// gitManifestLines is what flat manifests would hold for the fetch in
// dir: its commits times the files at HEAD. Zero when git cannot say,
// which leaves the import flat.
func gitManifestLines(ctx context.Context, env []string, dir string) int64 {
out, err := runGit(ctx, env, dir, "rev-list", "--all", "--count")
if err != nil {
return 0
}
commits, err := strconv.ParseInt(strings.TrimSpace(out), 10, 64)
if err != nil {
return 0
}
out, err = runGit(ctx, env, dir, "ls-tree", "-r", "--name-only", "-z", "HEAD")
if err != nil {
return 0
}
return commits * int64(strings.Count(out, "\x00"))
}
// pullGit is one pull of a Git mirror: a shallow fetch, then an import
// of what is new under the store's lease, deepening the fetch while zit
// says the history it has does not reach the store. An interrupted
// import is finished first. It answers whether the store changed, ref
// changes included.
func (rg *registrar) pullGit(name, store string, cli *zitcli.Client, guard func(func(lease.Token) error) error, upstream string) (bool, error) {
defer rg.gitLock(name)()
defer lockGitStore(store)()
recovered, err := rg.finishGitImport(name, store, cli, guard)
if err != nil {
return false, err
}
dir, done, err := rg.gitWorkspace(name)
if err != nil {
return false, err
}
defer done()
fctx, cancel := context.WithTimeout(context.Background(), gitFetchTimeout)
defer cancel()
depth := rg.gitFirstDepth
if depth == 0 {
depth = gitFirstDepth
}
for {
if err := rg.gitFetch(fctx, dir, upstream, depth); err != nil {
return false, err
}
var r zitcli.GitImport
err := guard(func(tok lease.Token) error {
ctx, cancel := context.WithTimeout(context.Background(), gitImportTimeout)
defer cancel()
var ierr error
r, ierr = cli.ImportGitInto(tok, ctx, dir, gitEnv(rg.gitProtocol()))
return ierr
})
if err != nil {
return false, err
}
if r.Retry == "deepen" && depth != 0 {
if depth *= 4; depth > gitMaxDepth {
depth = 0
}
continue
}
if err := r.Err(); err != nil {
return false, err
}
if r.Changed {
log.Printf("mirror %s: %d check-ins added (%s)", name, r.Added, r.Outcome)
}
return r.Changed || recovered, nil
}
}
// finishGitImport runs zit's recovery when the store holds an import a
// restart or a kill interrupted: until it does, zit refuses every read
// of the store. It answers whether recovery changed the store. The
// caller holds the repository's locks.
func (rg *registrar) finishGitImport(name, store string, cli *zitcli.Client, guard func(func(lease.Token) error) error) (bool, error) {
if !zitcli.GitImportPending(store) {
return false, nil
}
var r zitcli.GitImport
err := guard(func(tok lease.Token) error {
ctx, cancel := context.WithTimeout(context.Background(), gitImportTimeout)
defer cancel()
var rerr error
r, rerr = cli.RecoverGitImport(tok, ctx)
return rerr
})
if err == nil {
err = r.Err()
}
if err != nil {
return false, fmt.Errorf("finishing an interrupted import: %w", err)
}
log.Printf("mirror %s: finished an interrupted import (%s)", name, r.Outcome)
return r.Changed, nil
}
// recoverGitAtStart finishes an import interrupted by the last stop,
// so the repository reads again without waiting for its next pull.
func (rg *registrar) recoverGitAtStart(name, store string, cli *zitcli.Client, guard func(func(lease.Token) error) error) {
if !zitcli.GitImportPending(store) {
return
}
go func() {
defer rg.gitLock(name)()
defer lockGitStore(store)()
if _, err := rg.finishGitImport(name, store, cli, guard); err != nil {
log.Printf("mirror %s: %v", name, err)
}
}()
}