package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"log"
"os"
"path/filepath"
"strings"
"sync"
"time"
"craftdesign.group/quarry/internal/children"
"craftdesign.group/quarry/internal/lease"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
)
// A repository's mirror to a Git host (RFC 0022 rung 2): after trunk
// moves, the node holding the store's lease runs `zit git export
// --autopush` with the repository's deploy key, and records what
// happened in words.
//
// Triggered from afterFold, which is every way trunk moves (a push, an
// accepted landing, a merge the forge writes), and from a slow tick,
// which retries a failure after its backoff and makes a just-enabled
// mirror's first push. One run per repository at a time: a trigger
// while one runs makes exactly one more.
// gitMirrorTimeout bounds one run. A first export of a large history
// takes minutes; a hung ssh must not hold the repository's slot for
// ever.
const gitMirrorTimeout = 30 * time.Minute
// gitMirrorTick is how often the retry pass looks.
const gitMirrorTick = 30 * time.Second
// mirrorRuns is the per-repository run state of one kind of mirror
// push, shared by the Git and the Fossil mirror. The zero value is
// ready.
type mirrorRuns struct {
mu sync.Mutex
running map[string]bool
again map[string]bool
// run, when set, replaces the kind's own run: a test's counter.
run func(name string, at time.Time)
}
// kick runs name's push now with do, or once more after the run in
// flight. Never blocks the caller.
func (g *mirrorRuns) kick(name string, do func(name string, at time.Time)) {
g.mu.Lock()
if g.running == nil {
g.running, g.again = map[string]bool{}, map[string]bool{}
}
if g.running[name] {
g.again[name] = true
g.mu.Unlock()
return
}
g.running[name] = true
run := g.run
g.mu.Unlock()
if run == nil {
run = do
}
go func() {
for {
run(name, time.Now())
g.mu.Lock()
if !g.again[name] {
delete(g.running, name)
g.mu.Unlock()
return
}
delete(g.again, name)
g.mu.Unlock()
}
}()
}
// gitMirrors is the Git mirror's run state.
type gitMirrors struct {
mirrorRuns
// ssh, when set, replaces the ssh program in GIT_SSH_COMMAND: a
// test's stub that serves the push from a local bare repository.
ssh string
}
// gitMirrorKick runs name's mirror now, or once more after the run in
// flight. Never blocks the caller.
func (rg *registrar) gitMirrorKick(name string) {
if rg.odb == nil || !rg.jobs.store(name) {
return
}
rg.gitMirror.kick(name, rg.gitMirrorRun)
}
// gitMirrorRetry kicks every mirror this node runs; gitMirrorRun
// decides whether there is anything to do, which costs a row read.
func (rg *registrar) gitMirrorRetry() {
if rg.odb == nil {
return
}
repos, err := rg.odb.GitMirrorRepos()
if err != nil {
log.Printf("git mirror: list: %v", err)
return
}
for _, name := range repos {
rg.gitMirrorKick(name)
}
}
// gitMirrorRun is one pass for one repository: nothing when there is
// no mirror, it is paused, trunk is already pushed, or a failure is
// still backing off; otherwise one export and push.
func (rg *registrar) gitMirrorRun(name string, at time.Time) {
m, on, err := rg.odb.GitMirrorOf(name)
if err != nil || !on {
return
}
// Paused, not failed: making the repository public again, or
// lifting the takedown, resumes without anyone touching the
// mirror. The settings page says which (RFC 0022, Private).
//
// Read from the database, not the read gate's cache, and closed on
// error: the cache answers "public" for everything until its first
// load, and may be a few seconds behind another node's write.
if priv, err := rg.odb.PrivateRepos(); err != nil || priv[name] {
return
}
if _, held, err := rg.odb.TakedownOf(name); err != nil || held {
return
}
repo, ok := rg.deps.Repos.Get(name)
if !ok || repo.Zit == nil || repo.Index == nil || repo.Guard == nil {
return
}
tip, ok, err := repo.Index.TipOf("trunk")
if err != nil || !ok || tip.Tip == m.Pushed {
return
}
// The schedule is the store's, so the page says what this does.
if next, ok := oltp.MirrorNextTry(m.ErrAt, m.Fails); ok && at.Before(next) {
return
}
// Through the store's write lease: zit records the mirror in the
// store's config, and a node that loses the lease stops pushing.
var why string
var decided bool
if err := repo.Guard(func(tok lease.Token) error {
why, decided = rg.gitMirrorPush(name, m.PublicKey, func(ctx context.Context, mirror, remote, branch string, env []string) (string, error) {
return repo.Zit.GitExport(tok, ctx, mirror, remote, branch, env)
}, m.Remote, m.Branch)
// A push cut short by the lease going elsewhere says nothing
// about the mirror: the next holder pushes without a backoff.
if tok.Err() != nil {
decided = false
}
return nil
}); err != nil || !decided {
return
}
if why == "" {
if err := rg.odb.RecordGitMirrorPush(name, m.PublicKey, m.Remote, m.Branch, tip.Tip); err != nil {
log.Printf("git mirror: %s: record: %v", name, err)
}
return
}
if err := rg.odb.RecordGitMirrorFailure(name, m.PublicKey, m.Remote, m.Branch, why); err != nil {
log.Printf("git mirror: %s: record: %v", name, err)
}
}
// gitMirrorSweep removes the key files of pushes that never ended: a
// crash or a kill skips their removal. Run once at start, before any
// push, so a mirror that never pushes again does not keep one on disk.
func (rg *registrar) gitMirrorSweep() {
if rg.dir == "" {
return
}
if err := os.RemoveAll(filepath.Join(rg.dir, "gitmirror", "keys")); err != nil {
log.Printf("git mirror: sweep: %v", err)
}
}
// gitExport is zitcli.Client.GitExport, named so a run can be handed one.
type gitExport func(ctx context.Context, mirror, remote, branch string, env []string) (string, error)
// gitMirrorPush writes the key for the length of one push and runs it.
// why is "" on success, else the failure in words; decided is false
// when the push was stopped from outside (the forge stopping), which
// says nothing about the mirror and records nothing.
//
// The key file is 0600 in a fresh 0700 directory under the forge's own
// data directory, removed when the push ends, and swept at start
// (gitMirrorSweep) for a push that never ended. ssh reads no
// configuration but what is given here (-F /dev/null) and uses no
// agent, so nothing in the forge account's ssh setup can reach a
// mirror's host; git reads no global or system configuration, so no
// hook or helper there runs either. The host key is trusted on first
// use and kept, per forge, beside the mirrors.
//
// Each mirror exports into a Git repository of its own, named for its
// deploy key: turning a mirror off and on again, or purging a
// repository and making another of the same name, starts a fresh
// export rather than inheriting marks for history that no longer
// exists. Older exports of the same repository are removed.
func (rg *registrar) gitMirrorPush(name, publicKey string, export gitExport, remote, branch string) (why string, decided bool) {
home := filepath.Join(rg.dir, "gitmirror")
keys := filepath.Join(home, "keys")
if strings.ContainsAny(home, " \t\n'\"") {
// ssh splits UserKnownHostsFile on whitespace whatever the
// quoting, and a key path must survive the shell.
return "the forge's data directory has a space or quote in its path, so ssh cannot be pointed at it", true
}
if err := os.MkdirAll(keys, 0o700); err != nil {
log.Printf("git mirror: %s: %v", name, err)
return "the forge could not keep the mirror's Git repository", true
}
key, err := rg.odb.GitMirrorKey(name)
if err != nil {
log.Printf("git mirror: %s: key: %v", name, err)
return "the deploy key could not be read; stop mirroring and mirror again to make a new one", true
}
run, err := os.MkdirTemp(keys, "run-")
if err != nil {
log.Printf("git mirror: %s: %v", name, err)
return "the forge could not write the deploy key for the push", true
}
defer os.RemoveAll(run)
keyFile := filepath.Join(run, "key")
if err := os.WriteFile(keyFile, []byte(key), 0o600); err != nil {
log.Printf("git mirror: %s: %v", name, err)
return "the forge could not write the deploy key for the push", true
}
exports := filepath.Join(home, "exports", name)
if err := os.MkdirAll(exports, 0o700); err != nil {
log.Printf("git mirror: %s: %v", name, err)
return "the forge could not keep the mirror's Git repository", true
}
current := keyID(publicKey) + ".git"
if old, err := os.ReadDir(exports); err == nil {
for _, e := range old {
if e.Name() != current {
os.RemoveAll(filepath.Join(exports, e.Name()))
}
}
}
ssh := rg.gitMirror.ssh
if ssh == "" {
ssh = "ssh"
}
sshCmd := ssh + " -F /dev/null -i " + shellQuote(keyFile) +
" -o IdentitiesOnly=yes -o IdentityAgent=none -o BatchMode=yes -o ConnectTimeout=20" +
" -o ServerAliveInterval=15 -o ServerAliveCountMax=4" +
" -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile=" + shellQuote(filepath.Join(home, "known_hosts"))
env := []string{"GIT_SSH_COMMAND=" + sshCmd, "GIT_CONFIG_GLOBAL=/dev/null", "GIT_CONFIG_NOSYSTEM=1",
"GIT_TERMINAL_PROMPT=0", "SSH_AUTH_SOCK="}
ctx, cancel := context.WithTimeout(context.Background(), gitMirrorTimeout)
defer cancel()
out, err := export(ctx, filepath.Join(exports, current), remote, branch, env)
switch {
case err == nil:
return "", true
case errors.Is(err, ops.ErrStopping), children.Default.IsStopping():
// Refused at the spawn, or ended by the stop's own deadline:
// either way the stop, not the mirror, is why it failed.
return "", false
case ctx.Err() != nil:
return "the push did not finish within " + gitMirrorTimeout.String(), true
}
log.Printf("git mirror: %s: %v: %s", name, err, lastLines(out, 6))
return gitMirrorWords(out, branch), true
}
// keyID names a mirror's export by its deploy key.
func keyID(publicKey string) string {
sum := sha256.Sum256([]byte(publicKey))
return hex.EncodeToString(sum[:8])
}
// gitMirrorWords puts a refused push in words a repository admin can
// act on, from what git and ssh said.
func gitMirrorWords(out, branch string) string {
has := func(subs ...string) bool {
for _, s := range subs {
if strings.Contains(out, s) {
return true
}
}
return false
}
switch {
case has("Permission denied", "Key is invalid", "key is not authorized", "Write access to repository not granted",
"The requested URL returned error: 403"):
return "the host refused the deploy key: check it is added with write access"
case has("[rejected]", "non-fast-forward", "fetch first"):
return "the remote's " + branch + " has history this repository does not: it was not pushed"
case has("Host key verification failed", "REMOTE HOST IDENTIFICATION HAS CHANGED"):
return "the host's key is not the one it showed on the first push: it was not pushed"
case has("Repository not found", "does not appear to be a git repository", "repository not found"):
return "the host has no such repository, or the deploy key cannot see it"
case has("Could not resolve hostname", "Connection timed out", "Connection refused", "No route to host",
"Network is unreachable", "Operation timed out", "Connection closed by"):
return "the host could not be reached"
}
if l := strings.TrimRight(lastLines(out, 1), ". "); l != "" {
return "the push failed: " + l
}
return "the push failed"
}
// lastLines is the last n non-empty lines of out that are not git's
// hints, joined, with every control character a remote may have sent
// (a carriage return, an escape) written out as \xNN: the result goes to
// a log and into a stored error, and a remote must not be able to forge
// a line of either.
func lastLines(out string, n int) string {
var keep []string
for _, l := range strings.Split(out, "\n") {
l = strings.TrimSpace(l)
if l == "" || strings.HasPrefix(l, "hint:") {
continue
}
keep = append(keep, escapeControls(l))
}
if len(keep) > n {
keep = keep[len(keep)-n:]
}
return strings.Join(keep, " / ")
}
func escapeControls(s string) string {
var b strings.Builder
for _, r := range s {
if r < 0x20 || r == 0x7f || (r >= 0x80 && r < 0xa0) {
fmt.Fprintf(&b, `\x%02x`, r)
continue
}
b.WriteRune(r)
}
return b.String()
}
// shellQuote quotes s for the shell git runs GIT_SSH_COMMAND through.
func shellQuote(s string) string {
return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'"
}