package main
import (
"bytes"
"context"
"encoding/base64"
"errors"
"fmt"
"log"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"sync"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/children"
"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 an import asked to be
// flat stops and asks its owner to confirm sub-manifests, which stock
// Fossil cannot read. At the kernel's measured rate the limit is about
// half an hour of flat import.
const gitFlatLimit = 1_000_000_000
// errTooLargeForFlat is an import stopped at gitFlatLimit.
var errTooLargeForFlat = errors.New("too large for flat manifests")
// gitKeptFor is how long the fetch of an import stopped at gitFlatLimit
// is kept for its owner's answer: confirmed within it, the import starts
// from that fetch rather than fetching the whole history again (pkgsrc:
// 2.5 GB, the kernel: 6.5 GB). Past it the fetch is deleted, and a
// confirm fetches afresh.
const gitKeptFor = 24 * time.Hour
// gitKeptDir holds those fetches, one directory a repository, under the
// root the per-process workspaces live in but never swept with them.
const gitKeptDir = "kept"
// 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",
}
}
// withGitToken adds an access token to env for requests to upstream:
// a request header scoped to that URL, the way git sends a credential
// to one remote, so it is not in the URL, the fetch's arguments or its
// output, and no other address receives it. The user name is GitHub's
// for tokens; GitLab and Gitea take any name with a token as password.
// No token, no header: the fetch is anonymous.
func withGitToken(env []string, upstream, token string) []string {
if token == "" {
return env
}
n := 0
out := make([]string, 0, len(env)+2)
for _, kv := range env {
if v, ok := strings.CutPrefix(kv, "GIT_CONFIG_COUNT="); ok {
n, _ = strconv.Atoi(v)
continue
}
out = append(out, kv)
}
basic := base64.StdEncoding.EncodeToString([]byte("x-access-token:" + token))
i, j := strconv.Itoa(n), strconv.Itoa(n+1)
// No redirect is followed while a token is held: git moves its base
// URL to where a redirect points and sends the header there on every
// request after, so an upstream whose old address now redirects to
// another host would hand that host the token. A moved upstream
// fails instead, and says so.
return append(out,
"GIT_CONFIG_KEY_"+i+"=http."+upstream+".extraHeader",
"GIT_CONFIG_VALUE_"+i+"=Authorization: Basic "+basic,
"GIT_CONFIG_KEY_"+j+"=http.followRedirects",
"GIT_CONFIG_VALUE_"+j+"=false",
"GIT_CONFIG_COUNT="+strconv.Itoa(n+2))
}
// gitLoginHint says what to do about a fetch the upstream refused for
// want of a login: git's own words ("could not read Username ...
// terminal prompts disabled") name a terminal the forge does not have.
// Anything else is the error as it was.
func gitLoginHint(err error, hadToken bool) error {
msg := err.Error()
if !strings.Contains(msg, "could not read Username") && !strings.Contains(msg, "Authentication failed") &&
!strings.Contains(msg, "returned error: 403") && !strings.Contains(msg, "returned error: 401") {
return err
}
if hadToken {
return fmt.Errorf("the upstream refused the access token: it may have expired, or not grant read access to this repository's contents (%w)", err)
}
return fmt.Errorf("the upstream asked for a login: the repository is private, or does not exist. Start again with an access token (%w)", err)
}
// 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) && e.Name() != gitKeptDir {
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
}
rg.pruneKeptFetches(time.Now())
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
}
// keptFetch is where the fetch kept for name's answer lives, and the
// file in it naming the upstream it came from.
func (rg *registrar) keptFetch(name string) (dir, upstream string) {
dir = filepath.Join(rg.dir, "gitimport", gitKeptDir, name)
return dir, filepath.Join(dir, ".quarry-upstream")
}
// keepFetch moves the fetch at dir aside for name's answer, noting the
// upstream so it is only ever reused for the same one.
func (rg *registrar) keepFetch(dir, name, clone string) error {
kept, mark := rg.keptFetch(name)
if err := os.MkdirAll(filepath.Dir(kept), 0o700); err != nil {
return err
}
if err := os.RemoveAll(kept); err != nil {
return err
}
if err := os.Rename(dir, kept); err != nil {
return err
}
return os.WriteFile(mark, []byte(clone), 0o600)
}
// takeKeptFetch moves the fetch kept for name into dir, an empty
// workspace, when there is one from the same upstream and still within
// gitKeptFor, and reports whether it did. Any other kept fetch for name
// is deleted: it can never be used.
func (rg *registrar) takeKeptFetch(dir, name, clone string, now time.Time) bool {
kept, mark := rg.keptFetch(name)
info, err := os.Stat(mark)
if err != nil {
return false
}
up, err := os.ReadFile(mark)
if err != nil || string(up) != clone || now.Sub(info.ModTime()) > gitKeptFor {
_ = os.RemoveAll(kept)
return false
}
_ = os.Remove(mark)
if err := os.Remove(dir); err != nil {
return false
}
if err := os.Rename(kept, dir); err != nil {
_ = os.MkdirAll(dir, 0o700)
return false
}
return true
}
// pruneKeptFetches deletes the kept fetches past gitKeptFor.
func (rg *registrar) pruneKeptFetches(now time.Time) {
root := filepath.Join(rg.dir, "gitimport", gitKeptDir)
entries, _ := os.ReadDir(root)
for _, e := range entries {
_, mark := rg.keptFetch(e.Name())
if info, err := os.Stat(mark); err == nil && now.Sub(info.ModTime()) <= gitKeptFor {
continue
}
if err := os.RemoveAll(filepath.Join(root, e.Name())); err != nil {
log.Printf("kept git fetch %s: %v", e.Name(), err)
}
}
}
// 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, token string) error {
if rg.gitOnFetch != nil {
rg.gitOnFetch(depth)
}
env := withGitToken(gitEnv(rg.gitProtocol()), upstream, token)
// 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")
}
// Pull request heads let zit pair a squash merge with the branch it
// came from, so the timeline joins them (zit 0.8.1, zit#291). zit
// imports only the heads a squash pairs with; a host without such
// refs simply has none to fetch.
args = append(args, "--", upstream, "+refs/heads/*:refs/heads/*", "+refs/tags/*:refs/tags/*", "+refs/pull/*/head:refs/pull/*/head")
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...)
children.Confine(cmd)
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()
if spec.GitToken == "" && rg.takeKeptFetch(dir, spec.Name, spec.Clone, time.Now()) {
log.Printf("import %s: starting from the fetch kept for its answer", spec.Name)
} else {
fctx, cancel := context.WithTimeout(ctx, gitFetchTimeout)
defer cancel()
if err := rg.gitFetch(fctx, dir, spec.Clone, 0, spec.GitToken); err != nil {
return gitLoginHint(err, spec.GitToken != "")
}
}
// 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)
if !spec.ZitOnly {
limit := rg.gitFlatLimit
if limit == 0 {
limit = gitFlatLimit
}
if commits, files := gitHistorySize(ctx, gitEnv(rg.gitProtocol()), dir); commits*files > limit {
// A private upstream's fetch is never kept: it would be
// there for whoever next imports that name from that URL.
if spec.GitToken == "" {
if err := rg.keepFetch(dir, spec.Name, spec.Clone); err != nil {
log.Printf("import %s: keeping the fetch for the answer: %v", spec.Name, err)
}
}
return fmt.Errorf("%w: %s commits of up to %s files", errTooLargeForFlat, grouped(commits), grouped(files))
}
}
ictx, icancel := context.WithTimeout(ctx, gitFirstImportTimeout)
defer icancel()
r, err := zitcli.ImportGit(ictx, dir, spec.Path, gitEnv(rg.gitProtocol()), spec.ZitOnly)
if err == nil {
err = r.Err()
}
if err != nil {
if gitImportPublished(spec.Path) {
// zit writes the destination only once the history is
// whole, then warms its cache, which is advisory. A death
// in the warm (the kernel's ran out of memory, zit#293)
// leaves a complete store: serving it beats throwing away
// ten hours, and the cache warms on a later open.
log.Printf("import %s from Git %s: published, then %v; the store is whole and is kept", spec.Name, spec.Clone, err)
return nil
}
return err
}
log.Printf("import %s from Git %s: %d check-ins", spec.Name, spec.Clone, r.Added)
return nil
}
// gitImportPublished reports whether a first import got as far as
// publishing: zit makes the destination appear, store and import map
// together, only after the export, the check-ins and the refs are all
// written ("The import is transactional", zit import-git --help). Read
// from the files, not by asking zit: opening a store that large is the
// very step that can run out of memory.
func gitImportPublished(dest string) bool {
for _, f := range []string{".zit", ".zit-import-map"} {
if fi, err := os.Stat(filepath.Join(dest, f)); err != nil || !fi.Mode().IsRegular() {
return false
}
}
return true
}
// grouped is n with its thousands marked: 1,500,942.
func grouped(n int64) string {
s := strconv.FormatInt(n, 10)
for i := len(s) - 3; i > 0; i -= 3 {
s = s[:i] + "," + s[i:]
}
return s
}
// gitHistorySize is the fetch in dir's commits and the files at HEAD:
// flat manifests would hold their product in lines. Zeros when git
// cannot say, which leaves the import as it was asked for.
func gitHistorySize(ctx context.Context, env []string, dir string) (commits, files int64) {
out, err := runGit(ctx, env, dir, "rev-list", "--all", "--count")
if err != nil {
return 0, 0
}
commits, err = strconv.ParseInt(strings.TrimSpace(out), 10, 64)
if err != nil {
return 0, 0
}
out, err = runGit(ctx, env, dir, "ls-tree", "-r", "--name-only", "-z", "HEAD")
if err != nil {
return 0, 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
}
// The mirror's access token, for an upstream that is not public.
token := ""
if rg.odb != nil {
if token, err = rg.odb.MirrorGitToken(name); err != nil {
return false, fmt.Errorf("the mirror's access token: %w", err)
}
}
for {
if err := rg.gitFetch(fctx, dir, upstream, depth, token); err != nil {
return false, gitLoginHint(err, token != "")
}
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)
}
}()
}