Skip to content
quarry Artifact [edee362896]
Login

Artifact [edee362896]

Check-ins Using Download Hex

Artifact edee362896847bc491ac68656ea307b79b857a4a2780265118ca8c09a95d4585:

package main

import (
	"archive/tar"
	"bytes"
	"compress/gzip"
	"context"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"path"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unicode/utf8"

	"craftdesign.group/quarry/internal/api"
	"craftdesign.group/quarry/internal/ci"
	"craftdesign.group/quarry/internal/events"
	"craftdesign.group/quarry/internal/index"
	"craftdesign.group/quarry/internal/ingest"
	"craftdesign.group/quarry/internal/lease"
	"craftdesign.group/quarry/internal/oltp"
	"craftdesign.group/quarry/internal/ops"
	"craftdesign.group/quarry/internal/work"
	"craftdesign.group/quarry/internal/xfer"
	"craftdesign.group/quarry/internal/zitcli"

	"errors"
)

// registrar owns a served repository's whole lifetime: its index handle,
// its ingest pipeline, its sync backend, and the goroutine polling it.
//
// This was the body of serve's startup loop. It is a type now because a
// repository can be registered while the server runs, and the loop's
// shape did not survive that: it relied on `defer ix.Close()` reaching
// serve's return, and on a poll goroutine that never stops. Neither is
// available to an admin call, and both are leaks if a repository is ever
// removed.
type registrar struct {
	dir         string // derived-data root
	reposRoot   string // where repositories created through the API are placed
	node        string // this node's id, for durable consumer names
	interval    time.Duration
	ciCloneBase string

	ev         *events.Server
	leases     *lease.Manager
	deps       *api.Deps
	fronter    *xfer.Fronter
	ingestLane *work.Lane
	bulk       *work.Lane
	engine     *ci.Engine
	odb        *oltp.DB
	// ciTagsDefault is the fleet's verdict-tag default (RFC 0009); a
	// repository's own mode overrides it either way.
	ciTagsDefault bool

	// codes is shared with the auth verifier's cache, so a project code
	// learned here is the one the push gate checks.
	codes  map[string]string
	codeMu *sync.Mutex

	mu      sync.Mutex
	running map[string]*served

	privMu  sync.RWMutex
	private map[string]bool
}

// served is what a registration leaves behind that must be released.
type served struct {
	ix   *index.Index
	stop context.CancelFunc
	done chan struct{} // closed when the poll goroutine has exited
}

func (rg *registrar) track(name string, s *served) {
	rg.mu.Lock()
	defer rg.mu.Unlock()
	if rg.running == nil {
		rg.running = map[string]*served{}
	}
	rg.running[name] = s
}

// Register opens a repository and starts serving it. Safe to call while
// the server is running.
func (rg *registrar) Register(name, path string) error {
	// Startup binds the port only after every repository is registered,
	// so each repository's cost is the whole forge's downtime. Timed so
	// a slow one is nameable instead of merely suspected.
	regStart := time.Now()
	defer func() {
		if d := time.Since(regStart); d > 500*time.Millisecond {
			log.Printf("register %s: took %s", name, d.Round(time.Millisecond))
		}
	}()
	if _, exists := rg.deps.Repos.Get(name); exists {
		return fmt.Errorf("repository %q is already served: %w", name, ops.ErrInvalid)
	}
	if err := os.MkdirAll(filepath.Join(rg.dir, name), 0o755); err != nil {
		return err
	}
	ix, err := index.Open(filepath.Join(rg.dir, name, "index.db"))
	if err != nil {
		return err
	}
	cli := zitcli.New(path)
	// The lease keys on the store's project code, learned via the same
	// boundary-clean handshake user-add uses. Doing it before anything
	// is published means a path that is not a repository fails here,
	// with the index closed again, rather than half-registering.
	code, cerr := xfer.ProjectCode(path)
	if cerr != nil {
		ix.Close()
		// A path that is not a repository is a bad REQUEST, not a server
		// fault. Without this it surfaced as "internal error", which
		// tells an operator to read the server log for what is in fact
		// their own typo.
		return fmt.Errorf("%w: %s is not a zit repository: %v", ops.ErrInvalid, path, cerr)
	}
	rg.codeMu.Lock()
	rg.codes[name] = code
	rg.codeMu.Unlock()

	// Probed once here, not per render: the answer changes only when an
	// S-bearing artifact first arrives. A probe that fails leaves it
	// false, which withholds the fossil clone rather than offering one
	// that might not work.
	//
	// BOUNDED, because this runs before the server binds its port and
	// the cost is a function of the store's size, not a constant: the
	// note here used to say ~250ms, measured on small stores. On tcl it
	// is 116 SECONDS and on sqlite 16, so a fleet with either in it
	// could not answer a request for minutes after starting — the whole
	// forge held hostage by a detail about one clone URL. Past the
	// deadline the probe is simply an unanswered question, which the
	// false above already handles correctly.
	fossilClone := false
	probeCtx, cancelProbe := context.WithTimeout(context.Background(), 2*time.Second)
	if seen, err := cli.SeenSubmanifest(probeCtx); err == nil {
		fossilClone = !seen
	} else if probeCtx.Err() != nil {
		log.Printf("register %s: submanifest probe timed out; fossil clone withheld", name)
	} else {
		log.Printf("register %s: submanifest probe: %v", name, err)
	}
	cancelProbe()

	indexing := new(atomic.Bool)
	indexing.Store(true)
	// Bumped after every ingest round so memoised answers that depend
	// on the DAG — the leaf set below — recompute exactly when the DAG
	// can have changed, and never otherwise.
	// One warm pass per repository at a time. warmReads and indexTree
	// each spawn SEVERAL zit subprocesses — ls, ls-under, a cat per
	// file, an archive — and on a store tcl's size each takes seconds.
	// Launched unguarded from every round that moved the tip, the next
	// round started another set before the last had finished and they
	// stacked: 43 zit children on the demo fleet, three of them the
	// same ls against the same hash, with load average 546 on 14 cores.
	// A warm pass already running IS the warm pass; a second one is
	// duplicate work that makes the first one slower.
	var warming, indexingTree atomic.Bool
	ingestGen := new(atomic.Uint64)
	entry := &api.Repo{
		FossilClone: fossilClone,
		Index:       ix, Zit: cli, Code: code,
		Indexing: indexing.Load,
		Guard: func(fn func() error) error {
			err := rg.leases.Guard(code, fn)
			if errors.Is(err, lease.ErrNotHolder) || errors.Is(err, lease.ErrUnavailable) {
				return fmt.Errorf("%w: %v", ops.ErrUnavailable, err)
			}
			return err
		},
	}
	// Leaf status is zit's answer, memoised per ingest: the set only
	// changes when a check-in arrives.
	//
	// The memo is WARMED IN THE BACKGROUND rather than filled by the
	// first reader. That call is not ~10ms on a large store as this
	// comment used to claim — measured on sqlite it is 6.4 SECONDS, and
	// making one unlucky request pay it put the repository home at
	// 10.8s cold against 0.5s warm. It was invisible in the footer too,
	// because a detached context carries no timer, so the time landed
	// in "ops" with nothing to point at.
	//
	// A reader arriving before the warm finishes gets no leaf marks
	// rather than a six-second wait: the marks are decoration on a
	// timeline that is otherwise complete, and a page that arrives
	// without them is better than one that does not arrive.
	var leafMu sync.Mutex
	var leafSet map[string]bool
	var leafErr error
	var leafGen uint64
	var leafWarming bool
	warmLeaves := func(gen uint64) {
		set, err := cli.Leaves(context.Background())
		leafMu.Lock()
		leafSet, leafErr, leafGen, leafWarming = set, err, gen, false
		leafMu.Unlock()
	}
	entry.Leaves = func() (map[string]bool, error) {
		gen := ingestGen.Load()
		leafMu.Lock()
		defer leafMu.Unlock()
		if leafSet != nil && leafGen == gen {
			return leafSet, leafErr
		}
		if !leafWarming {
			leafWarming = true
			go warmLeaves(gen)
		}
		// Stale is better than slow: an older set still marks almost
		// every leaf correctly, and nil simply draws none.
		return leafSet, leafErr
	}
	// Warm at registration so the first reader of a repository does not
	// find it cold either.
	go func() {
		leafMu.Lock()
		if leafWarming {
			leafMu.Unlock()
			return
		}
		leafWarming = true
		gen := ingestGen.Load()
		leafMu.Unlock()
		warmLeaves(gen)
	}()
	// The default branch is zit's resolution of "trunk" — fossil's
	// main-branch setting and tag propagation live there, so a repo
	// that renamed trunk to main answers "main". Memoised per ingest
	// generation like the leaf set: it changes only when the DAG does.
	var dbMu sync.Mutex
	var dbName, dbTip string
	var dbGen uint64
	dbGen = ^uint64(0)
	entry.DefaultBranch = func() (string, string) {
		gen := ingestGen.Load()
		dbMu.Lock()
		defer dbMu.Unlock()
		if dbGen == gen {
			return dbName, dbTip
		}
		tip, name, err := cli.Info(context.Background(), "trunk")
		if err != nil {
			return "", "" // caller falls back to the index heuristic
		}
		dbName, dbTip, dbGen = name, tip, gen
		return dbName, dbTip
	}
	entry.Mirror = func(url string) (bool, error) {
		var changed bool
		err := entry.Guard(func() error {
			var perr error
			changed, perr = cli.Pull(url)
			return perr
		})
		return changed, err
	}
	pipe := ingest.NewPipeline(cli, rg.ev, ix, name, rg.node)
	// In-process reindex: through the same lane as every other ingest,
	// so a fleet-wide repair is staged at --ingest-limit stores at a
	// time rather than a stampede; the indexing flag makes the fleet
	// page say what is happening; the generation bump drops every memo
	// derived from the shape being rebuilt.
	entry.Reindex = func() (int, error) {
		var n int
		indexing.Store(true)
		defer indexing.Store(false)
		err := rg.ingestLane.Do(context.Background(), func() error {
			var rerr error
			n, rerr = pipe.Reindex()
			return rerr
		})
		ingestGen.Add(1)
		// Reset drops every derived table, and the doc rows are one of
		// them. Without this a rebuild leaves documentation silently
		// unsearchable until the next commit happens to move the tip —
		// which on a mirror of a quiet project is never.
		if err == nil {
			if indexingTree.CompareAndSwap(false, true) {
				go func() {
					defer indexingTree.Store(false)
					indexTree(cli, ix)
				}()
			}
		}
		return n, err
	}
	var warmOnce sync.Once
	entry.Sync = func() (int, error) {
		n, err := pipe.Sync()
		// Only a round that actually ingested something can have moved
		// the DAG, so only that invalidates the memos hanging off it.
		// Bumping on every round made the poll interval the memo's
		// lifetime: with a two-second poll, essentially every page load
		// paid the cold path — 84ms against 9ms on fossil.
		if n > 0 {
			ingestGen.Add(1)
		}
		// Warm the reader's path before a reader pays for it: the memo
		// caches are per-process and keyed on the tip, so both a restart
		// and a new check-in leave the next files-page visitor holding
		// four cold zit subprocesses — 5.6 seconds on a store tcl's
		// size. After the round that moved the tip (and once at boot),
		// the same reads run here instead, in the background.
		if err == nil {
			fresh := n > 0
			warmOnce.Do(func() { fresh = true })
			if fresh {
				// CompareAndSwap claims the slot: false means nobody is
				// warming, and the winner clears it on the way out.
				if warming.CompareAndSwap(false, true) {
					go func() {
						defer warming.Store(false)
						warmReads(cli, ix)
					}()
				}
				if indexingTree.CompareAndSwap(false, true) {
					go func() {
						defer indexingTree.Store(false)
						indexTree(cli, ix)
					}()
				}
			}
		}
		return n, err
	}
	entry.Health = func() api.RepoHealth {
		s := pipe.Stats()
		h := api.RepoHealth{
			RoundSeconds: s.LastRound.Seconds(),
			LastCount:    s.LastCount,
			Total:        s.Total,
			Fallbacks:    s.Fallbacks,
			Gaps:         s.Gaps,
			Stalled:      s.Stalled,
			RetryAt:      retryAt(s.RetryAt),
			LastErr:      s.LastErr,
			Failures:     s.Failures,
		}
		if !s.LastOK.IsZero() {
			h.LastOK = s.LastOK.Format(time.RFC3339)
		}
		if !s.LastErrAt.IsZero() {
			h.LastErrAt = s.LastErrAt.Format(time.RFC3339)
		}
		return h
	}

	rg.fronter.Register(name, path)
	// Published complete or not at all.
	rg.deps.Repos.Put(name, entry)

	ctx, stop := context.WithCancel(context.Background())
	done := make(chan struct{})
	rg.track(name, &served{ix: ix, stop: stop, done: done})
	go rg.poll(ctx, done, name, path, cli, ix, entry.Sync, indexing)
	return nil
}

// poll runs the store's initial ingest and then its round loop until the
// context is cancelled.

// retryAt formats the stall's next attempt, empty when not stalled.
func retryAt(t time.Time) string {
	if t.IsZero() {
		return ""
	}
	return t.UTC().Format(time.RFC3339)
}

// sync is entry.Sync — the round PLUS the generation bump that lets
// memoised DAG answers know they may be stale. The poll must not call
// the pipeline directly, or those memos would outlive the ingest that
// invalidated them.
func (rg *registrar) poll(ctx context.Context, done chan struct{}, name, path string,
	cli *zitcli.Client, ix *index.Index, sync func() (int, error), indexing *atomic.Bool) {
	defer close(done)

	start := time.Now()
	var n int
	err := rg.ingestLane.Do(ctx, func() error {
		var serr error
		n, serr = sync()
		return serr
	})
	indexing.Store(false)
	if err != nil {
		log.Printf("initial ingest of %s: %v", name, err)
	} else {
		log.Printf("ingested %d artifacts from %s (%s) in %s",
			n, name, path, time.Since(start).Round(time.Second))
	}

	// warmedTip is the version whose listing has already been warmed,
	// warmingTip whether a warm is in flight. Both belong to this poll
	// loop: one repository, one warm at a time, and never twice for the
	// same tip.
	var warmingTip atomic.Bool
	var warmedTip atomic.Value
	warmedTip.Store("")

	// time.Tick leaks its ticker; a Ticker can be stopped, which matters
	// now that a repository can be unregistered.
	tick := time.NewTicker(rg.interval)
	defer tick.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
		}
		if err := rg.ingestLane.Do(ctx, func() error {
			_, serr := sync()
			return serr
		}); err != nil {
			if ctx.Err() != nil {
				return
			}
			// The stall already announced itself once, with the cause;
			// repeating it every poll interval buries the log line that
			// matters under hundreds that do not.
			if !errors.Is(err, ingest.ErrStalled) {
				log.Printf("ingest %s: %v", name, err)
			}
			continue
		}
		// Warm the tip's listing so the repo home is never the request
		// that pays for it: `ls --long` costs ~4ms per file, which is
		// 8.5s on a 2,198-path tree. The listing is immutable per
		// version, so it is computed once and reused until the tip moves.
		//
		// ONLY when the tip has actually moved, and only one at a time.
		// This fired on EVERY poll tick as a bare goroutine whose result
		// was discarded — so on a store where the call takes seconds the
		// next tick started another before the last finished. Measured
		// on the demo fleet: 134 concurrent `zit ls --long`, all against
		// the SAME hash, load average 546 on 14 cores. A listing that is
		// immutable per version needs computing once per version, and
		// the memo behind it already knows that; what was missing was
		// anyone asking whether the work was already in flight.
		if b, ok, err := ix.TipOf("trunk"); err == nil && ok {
			if warmedTip.Load() != b.Tip && warmingTip.CompareAndSwap(false, true) {
				tip := b.Tip
				go func() {
					defer warmingTip.Store(false)
					if _, err := cli.Ls(context.Background(), tip); err == nil {
						warmedTip.Store(tip)
					}
				}()
			}
		}
		// CI looks right after a fold, which is the moment a new
		// check-in becomes visible to quarry whatever brought it in.
		if rg.engine.Enabled() {
			if err := ci.Trigger(rg.engine, ciStore{db: rg.odb}, name,
				ciRepo{ix: ix, zit: cli}, rg.ciCloneBase+name); err != nil {
				log.Printf("ci: %s: %v", name, err)
			}
		}
		rg.gateSweep(name, ix)
		rg.carryForward(name, ix)
	}
}

// gateSweep is the bypass tripwire (RFC 0007): a protected branch that
// moved without a forge accept behind it is named, loudly, within a
// poll tick — because a stock zit client can accept against the store
// and the store cannot refuse it; noticing is the forge's half.
func (rg *registrar) gateSweep(name string, ix *index.Index) {
	if rg.odb == nil {
		return
	}
	rules, err := rg.odb.Protections(name)
	if err != nil {
		return
	}
	for _, rule := range rules {
		if strings.ContainsAny(rule.Branch, "*?[") {
			continue // a glob names no single tip to watch; a rung for later
		}
		b, ok, err := ix.TipOf(rule.Branch)
		if err != nil || !ok {
			continue
		}
		watch, seen := rg.odb.GateWatch(name, rule.Branch)
		if !seen {
			// First sight: baseline at the tip. Retroactively flagging
			// history would bury the signal under old noise.
			_ = rg.odb.SetGateWatch(name, rule.Branch, b.Tip)
			continue
		}
		if watch == b.Tip {
			continue
		}
		fresh, err := ix.CheckinsAfter(watch, 50)
		if err != nil {
			continue
		}
		for _, e := range fresh {
			if e.Branch != rule.Branch {
				continue
			}
			if ok, _ := rg.odb.ForgeAccepted(name, e.Hash); ok {
				continue
			}
			if err := rg.odb.AddGateBypass(name, rule.Branch, e.Hash); err == nil {
				log.Printf("gate: %s %s moved past its checks at %.12s — no forge accept behind it", name, rule.Branch, e.Hash)
			}
		}
		_ = rg.odb.SetGateWatch(name, rule.Branch, b.Tip)
	}
}

// carryForward writes a premerge pair's verdict onto the merge
// check-in an accept created from that pair (RFC 0009 §premerge): the
// portable record says this merge was tested before it existed. Runs
// after every fold; a merge already carrying a verdict is skipped, so
// the write happens once.
func (rg *registrar) carryForward(name string, ix *index.Index) {
	switch rg.odb.CITagsMode(name) {
	case "off":
		return
	case "on":
	default:
		if !rg.ciTagsDefault {
			return
		}
	}
	repo, ok := rg.deps.Repos.Get(name)
	if !ok || repo.Index == nil {
		return
	}
	landed, err := ix.Landings("accepted", 10)
	if err != nil {
		return
	}
	ids := make([]string, 0, len(landed))
	for _, l := range landed {
		ids = append(ids, l.ID)
	}
	merges, err := ix.MergesForLandings(ids)
	if err != nil {
		return
	}
	for _, l := range landed {
		m := merges[l.ID]
		if m.Checkin == "" {
			continue
		}
		if v, verr := ix.VerdictOf(m.Checkin); verr != nil || v != "" {
			continue
		}
		parents, perr := ix.ParentsOf(m.Checkin)
		if perr != nil || len(parents) != 2 {
			continue
		}
		// The tested pair was branchTip:targetTip; parent order is not
		// recorded, so ask both spellings.
		verdict := rg.deps.PairVerdict(name, "pair:"+parents[0]+":"+parents[1], l.Target)
		if verdict == "" {
			verdict = rg.deps.PairVerdict(name, "pair:"+parents[1]+":"+parents[0], l.Target)
		}
		if verdict == "" {
			continue
		}
		if err := repo.WriteTag(verdict, m.Checkin, "premerge"); err != nil {
			log.Printf("ci: carry-forward %s %.12s: %v", name, m.Checkin, err)
			continue
		}
		log.Printf("ci: %s %.12s %s (carried from premerge)", name, m.Checkin, verdict)
	}
}

// TriggerCI launches a repository's pipelines at a check-in — the
// re-run path (RFC 0009 §triggers). Empty checkin means the tip.
func (rg *registrar) TriggerCI(name, checkin string) error {
	if !rg.engine.Enabled() {
		return fmt.Errorf("no CI engine configured")
	}
	repo, ok := rg.deps.Repos.Get(name)
	if !ok || repo.Index == nil || repo.Zit == nil {
		return fmt.Errorf("no repository %q", name)
	}
	return ci.RunAt(rg.engine, ciStore{db: rg.odb}, name,
		ciRepo{ix: repo.Index, zit: repo.Zit}, rg.ciCloneBase+name, checkin)
}

// Unregister stops serving a repository and releases what it held.
//
// Order matters: drop it from the registry FIRST so no new request can
// find it, then stop the poll loop and wait for it to exit, then close
// the index. Closing while a fold is mid-transaction is how a "removed"
// repository takes the process down with it.
func (rg *registrar) Unregister(name string) error {
	if _, ok := rg.deps.Repos.Get(name); !ok {
		return fmt.Errorf("no repository %q: %w", name, ops.ErrNotFound)
	}
	rg.deps.Repos.Delete(name)
	rg.fronter.Unregister(name)

	rg.mu.Lock()
	s := rg.running[name]
	delete(rg.running, name)
	rg.mu.Unlock()

	if s != nil {
		s.stop()
		<-s.done
		s.ix.Close()
	}
	rg.codeMu.Lock()
	delete(rg.codes, name)
	rg.codeMu.Unlock()
	return nil
}

// Add creates or adopts a repository, records it, and starts serving it.
//
// The order is deliberate: make the store, then persist, then serve. A
// recorded repository that failed to open is visible and fixable; one
// that serves without being recorded vanishes at the next restart and
// leaves no trace of why.
func (rg *registrar) Add(spec api.RepoSpec) (string, error) {
	if rg.odb == nil {
		return "", fmt.Errorf("no forge database: %w", ops.ErrUnavailable)
	}
	if _, exists := rg.deps.Repos.Get(spec.Name); exists {
		return "", fmt.Errorf("repository %q is already served: %w", spec.Name, ops.ErrInvalid)
	}
	// The name is validated BEFORE it is joined to a path, because that
	// is the only thing standing between a name and the filesystem.
	if !oltp.RepoName.MatchString(spec.Name) {
		return "", fmt.Errorf("%w: repository name %q: use letters, digits, dot, dash, underscore",
			ops.ErrInvalid, spec.Name)
	}
	if spec.Path == "" {
		if rg.reposRoot == "" {
			return "", fmt.Errorf("%w: this node has no repositories root, so path is required "+
				"(start it with --repos-root DIR)", ops.ErrInvalid)
		}
		if !spec.Create && spec.Clone == "" {
			return "", fmt.Errorf("%w: path is required when adopting an existing store", ops.ErrInvalid)
		}
		spec.Path = filepath.Join(rg.reposRoot, spec.Name)
	}

	// Making a store is the one step that leaves something behind if a
	// later one fails, so it is undone explicitly rather than left for
	// an operator to find.
	made := false
	switch {
	case spec.Create:
		if err := rg.mustBeEmpty(spec.Path); err != nil {
			return "", err
		}
		if err := zitcli.Init(context.Background(), spec.Path); err != nil {
			return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err)
		}
		made = true
		// Seeded BEFORE the repository is registered, so the ingest
		// round that follows registration sees the first check-in
		// already there rather than racing it.
		if files := seedFiles(spec.Name, spec.Description, spec.License, spec.Ignore); spec.Readme || spec.License != "" || spec.Ignore {
			if !spec.Readme {
				delete(files, "README.md")
			}
			if err := zitcli.SeedFiles(context.Background(), spec.Path,
				spec.Author, "Initial check-in", files); err != nil {
				os.RemoveAll(spec.Path)
				return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err)
			}
		}
	case spec.Clone != "":
		if err := rg.mustBeEmpty(spec.Path); err != nil {
			return "", err
		}
		// An import is REGISTERED FIRST and copied afterwards. Cloning
		// inline meant the repository did not exist anywhere until the
		// last byte arrived: no row, no page, no progress, and an
		// operator watching the fleet saw nothing at all for the
		// minutes a large repository takes. Now the row is written
		// immediately in state Cloning, this call returns, and the copy
		// runs behind it — so the repository can be looked at while it
		// fills, and a failure has somewhere to be reported.
		if err := rg.odb.AddImportingRepo(spec.Name, spec.Path, spec.Description,
			spec.Visibility, spec.Clone); err != nil {
			return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err)
		}
		if spec.Mirror {
			rg.recordMirror(spec)
		}
		go rg.importClone(spec)
		return spec.Path, nil
	}

	undo := func() {
		if made {
			os.RemoveAll(spec.Path)
		}
	}
	if err := rg.odb.AddRepo(spec.Name, spec.Path, spec.Description, spec.Visibility); err != nil {
		undo()
		return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err)
	}
	if err := rg.Register(spec.Name, spec.Path); err != nil {
		if derr := rg.odb.DeleteRepo(spec.Name); derr != nil {
			log.Printf("repo.add %s: rollback failed: %v", spec.Name, derr)
		}
		undo()
		return "", err
	}
	if spec.Mirror && spec.Clone != "" {
		rg.recordMirror(spec)
	}
	return spec.Path, nil
}

// recordMirror attaches the upstream a clone came from. A failed mirror
// record is not worth unwinding a working repository over — it is one
// admin call to fix, and the repository is already serving.
func (rg *registrar) recordMirror(spec api.RepoSpec) {
	every := time.Duration(spec.MirrorEvery) * time.Second
	if every == 0 {
		every = 24 * time.Hour
	}
	if err := rg.odb.SetMirror(spec.Name, spec.Clone, every); err != nil {
		log.Printf("repo.add %s: mirror not recorded: %v", spec.Name, err)
	}
}

// importClone copies the upstream into an already-registered row, then
// moves it from Cloning to Ready. Every exit updates the row: a failure
// that only logged would leave a repository claiming to be importing
// with nothing importing it.
func (rg *registrar) importClone(spec api.RepoSpec) {
	// The bulk lane, as before: a clone is minutes of network and
	// memory, and admins asking for three at once should queue rather
	// than run unbounded.
	err := rg.bulk.Do(context.Background(), func() error {
		return zitcli.Clone(context.Background(), spec.Clone, spec.Path)
	})
	if err != nil {
		why := err.Error()
		if errors.Is(err, work.ErrBusy) {
			why = "too many imports in flight; nothing was copied"
		}
		os.RemoveAll(spec.Path)
		if serr := rg.odb.SetRepoState(spec.Name, oltp.Failed, why); serr != nil {
			log.Printf("import %s: recording failure: %v", spec.Name, serr)
		}
		log.Printf("import %s from %s: %v", spec.Name, spec.Clone, err)
		return
	}
	if err := rg.Register(spec.Name, spec.Path); err != nil {
		os.RemoveAll(spec.Path)
		if serr := rg.odb.SetRepoState(spec.Name, oltp.Failed, err.Error()); serr != nil {
			log.Printf("import %s: recording failure: %v", spec.Name, serr)
		}
		log.Printf("import %s: registering: %v", spec.Name, err)
		return
	}
	if err := rg.odb.SetRepoState(spec.Name, oltp.Ready, ""); err != nil {
		log.Printf("import %s: marking ready: %v", spec.Name, err)
	}
	log.Printf("import %s from %s: complete", spec.Name, spec.Clone)
}

// Imports reports what is copying now and what failed copying. Bytes
// are measured from the store on disk rather than tracked in memory, so
// the number survives a reader arriving mid-import and needs no
// bookkeeping the copy would have to keep up to date.
func (rg *registrar) Imports() ([]api.RepoImport, error) {
	if rg.odb == nil {
		return nil, nil
	}
	rows, err := rg.odb.Repos()
	if err != nil {
		return nil, err
	}
	var out []api.RepoImport
	for _, r := range rows {
		if r.State == oltp.Ready {
			continue
		}
		out = append(out, api.RepoImport{
			Name: r.Name, URL: r.ImportURL, State: r.State,
			Error: r.ImportError, Started: r.Created, Bytes: dirBytes(r.Path),
		})
	}
	return out, nil
}

// dirBytes totals a store's files. A missing directory is zero, not an
// error: a failed import has already had its partial store removed.
func dirBytes(path string) int64 {
	if path == "" {
		return 0
	}
	var total int64
	filepath.WalkDir(path, func(_ string, e os.DirEntry, err error) error {
		if err != nil || e.IsDir() {
			return nil //nolint:nilerr // an unreadable entry contributes nothing
		}
		if info, ierr := e.Info(); ierr == nil {
			total += info.Size()
		}
		return nil
	})
	return total
}

// ReconcileImports settles rows left mid-import by a restart. A clone
// runs as a CHILD of this server, so stopping the server kills it and
// the row would otherwise claim to be cloning forever — which is
// exactly what a deploy during an import produced. Marking them failed
// is the honest state: the copy is not running, and the partial store
// is removed so a retry starts clean.
func (rg *registrar) ReconcileImports() {
	if rg.odb == nil {
		return
	}
	names, err := rg.odb.ImportingRepos()
	if err != nil {
		log.Printf("imports: listing interrupted: %v", err)
		return
	}
	for _, name := range names {
		row, ok, err := rg.odb.RepoByName(name)
		if err != nil || !ok {
			continue
		}
		if row.Path != "" {
			os.RemoveAll(row.Path)
		}
		if err := rg.odb.SetRepoState(name, oltp.Failed,
			"interrupted by a server restart before the copy finished"); err != nil {
			log.Printf("imports: settling %s: %v", name, err)
			continue
		}
		log.Printf("imports: %s was interrupted by a restart; marked failed", name)
	}
}

// mustBeEmpty refuses to create or clone over existing content. Both zit
// verbs create the directory themselves, so anything already there is
// somebody else's, and the failure mode of guessing wrong is destroying
// it on the rollback path.
func (rg *registrar) mustBeEmpty(path string) error {
	if path == "" {
		return fmt.Errorf("%w: empty path", ops.ErrInvalid)
	}
	entries, err := os.ReadDir(path)
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("%w: %v", ops.ErrInvalid, err)
	}
	if len(entries) > 0 {
		return fmt.Errorf("%w: %s already exists and is not empty", ops.ErrInvalid, path)
	}
	return nil
}

// Remove stops serving a repository and forgets it. The store on disk
// and its derived index are left alone: "stop serving this" and
// "destroy this" are different intentions and only one is reversible.
func (rg *registrar) Remove(name string) error {
	// A repository whose import failed has a ROW but was never served,
	// so there is nothing to unregister — and refusing on that basis
	// would make a failed import impossible to clear from anywhere but
	// the database. Forgetting a registration that does not exist is
	// the outcome the caller asked for either way.
	if err := rg.Unregister(name); err != nil && !errors.Is(err, ops.ErrNotFound) {
		return err
	}
	if rg.odb != nil {
		if err := rg.odb.DeleteRepo(name); err != nil {
			return err
		}
	}
	return nil
}

// CloseAll releases every registration — serve's shutdown path.
func (rg *registrar) CloseAll() {
	for _, name := range rg.deps.Repos.Names() {
		if err := rg.Unregister(name); err != nil {
			log.Printf("unregister %s: %v", name, err)
		}
	}
}

// accountStore adapts forge OLTP to the ops layer's account seam. The
// role strings cross as text because that is the API's shape; parsing
// them here keeps oltp.Role out of the op surface.
type accountStore struct{ db *oltp.DB }

func (a accountStore) Accounts() ([]api.Account, error) {
	rows, err := a.db.Accounts()
	if err != nil {
		return nil, err
	}
	out := make([]api.Account, 0, len(rows))
	for _, r := range rows {
		out = append(out, api.Account{Name: r.Name, Admin: r.Admin, Created: r.Created})
	}
	return out, nil
}

func (a accountStore) AddAccount(name, password string, admin bool) error {
	return a.db.AddAccount(name, password, admin)
}

func (a accountStore) SetPassword(name, password string) error {
	return a.db.SetPassword(name, password)
}

func (a accountStore) DeleteAccount(name string) error { return a.db.DeleteAccount(name) }

func (a accountStore) Grant(account, repo, role string) error {
	r, err := oltp.ParseRole(role)
	if err != nil {
		return err
	}
	return a.db.Grant(account, repo, r)
}

func (a accountStore) Revoke(account, repo string) error { return a.db.Revoke(account, repo) }

func (a accountStore) GrantsFor(account string) (map[string]string, error) {
	grants, err := a.db.GrantsFor(account)
	if err != nil {
		return nil, err
	}
	out := make(map[string]string, len(grants))
	for repo, role := range grants {
		out[repo] = role.String()
	}
	// The "" key is the site-admin marker the dashboard reads ("a site
	// admin's forge IS the fleet"). RoleOf already grants site admins
	// RepoAdmin everywhere; without this map-shaped statement of the
	// same fact, an admin with no explicit rows saw an empty dashboard.
	if acct, found := a.db.AccountByName(account); found && acct.Admin {
		out[""] = oltp.RepoAdmin.String()
	}
	return out, nil
}

// GuardGrant asks the RFC 0008 guard about a grant before the API
// makes it. Callers hold s, so the guard warns rather than refuses —
// the last-admin removal happens, and the caller is told.
func (a accountStore) GuardGrant(actor, account, repo, role string) (string, error) {
	r, err := oltp.ParseRole(role)
	if err != nil {
		return "", err
	}
	return a.db.GuardCaps(actor, account, repo, r.Caps())
}

// GuardRevoke is GuardGrant for taking the string away.
func (a accountStore) GuardRevoke(actor, account, repo string) (string, error) {
	return a.db.GuardCaps(actor, account, repo, "")
}

// CapsAt is account.explain's seam. An unknown account is an error,
// not an empty string: the op explains people, not names.
func (a accountStore) CapsAt(account, repo string) (string, string, error) {
	if _, found := a.db.AccountByName(account); !found {
		return "", "", fmt.Errorf("no such account %q", account)
	}
	stored, err := a.db.CapsOf(account, repo)
	if err != nil {
		return "", "", err
	}
	return stored, string(a.db.EffectiveCaps(account, repo)), nil
}

// warmSem bounds concurrent warms fleet-wide: a restart syncs every
// store at once, and 23 parallel zit processes over the biggest repos
// is a memory spike with no reader waiting on it.
var warmSem = make(chan struct{}, 2)

// warmReads performs the reads a repository's files page performs, so
// the per-process memos (keyed on the immutable tip hash) are hot when
// a reader arrives. Failures are irrelevant — the reader's own request
// takes the cold path, exactly as if this had never run.
func warmReads(cli *zitcli.Client, ix *index.Index) {
	if cli == nil || ix == nil {
		return
	}
	warmSem <- struct{}{}
	defer func() { <-warmSem }()
	tip, ok, err := ix.TipOf(ix.DefaultBranch())
	if err != nil || !ok {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	defer cancel()
	files, err := cli.Ls(ctx, tip.Tip)
	if err != nil {
		return
	}
	cli.LsUnder(ctx, tip.Tip, "")
	for _, f := range files {
		base := strings.ToLower(path.Base(f.Name))
		if strings.HasPrefix(base, "readme") || strings.HasPrefix(base, "license") ||
			strings.HasPrefix(base, "copying") || f.Name == ".quarry/repo.json" {
			cli.Cat(ctx, f.Name, tip.Tip)
		}
	}
}

// indexTree files the default branch tip's FILES into the search
// index: fossil's search scope "d" for documentation, plus source,
// which fossil does not search at all.
//
// quarry rendered both as first-class surfaces — the docs tab, the file
// browser — and could find neither. A reader could open
// www/quickstart.wiki and get nothing searching for it.
//
// One tarball, not a read per file. `cat` per path costs a subprocess
// each: 2,199 files on sqlite is ~31 seconds and 2,199 store opens,
// against 2.2 seconds for one archive. The tarball is the whole tree in
// one pass, which is exactly the shape of the question.
//
// Asynchronous, on the triggers a rebuild and a moved tip already
// carry. Not through dagnats: dagnats is optional — a forge without one
// "simply has no CI rather than a broken one" — and search must not
// inherit that. This is local derivation from the store, the same class
// of work as the fold, which Reindex already routes around the bus on
// purpose.
func indexTree(cli *zitcli.Client, ix *index.Index) {
	if cli == nil || ix == nil {
		return
	}
	warmSem <- struct{}{}
	defer func() { <-warmSem }()
	tip, ok, err := ix.TipOf(ix.DefaultBranch())
	if err != nil || !ok {
		return
	}
	// The tip has not moved since the last pass: the rows already
	// describe it. This is what makes the boot trigger cheap on a
	// restart that ingested nothing.
	if at, verr := ix.TreeVersion(); verr == nil && at == tip.Tip {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
	defer cancel()

	tmp, err := os.CreateTemp("", "quarry-tree-*.tar.gz")
	if err != nil {
		return
	}
	defer os.Remove(tmp.Name())
	tmp.Close()
	if err := cli.ArchiveTo(ctx, "tar", tip.Tip, "", tmp.Name(), treeArchiveMaxBytes); err != nil {
		log.Printf("tree index: archive %.12s: %v", tip.Tip, err)
		return
	}
	entries, skipped, err := readTreeArchive(tmp.Name())
	if err != nil {
		log.Printf("tree index: read %.12s: %v", tip.Tip, err)
		return
	}
	if err := ix.ReplaceTree(tip.Tip, entries); err != nil {
		log.Printf("tree index: %v", err)
		return
	}
	var docs int
	for _, e := range entries {
		if e.Doc {
			docs++
		}
	}
	log.Printf("tree index: %d files (%d documentation) at %.12s, %d skipped",
		len(entries), docs, tip.Tip, skipped)
}

// readTreeArchive turns the tip tarball into index rows, skipping what
// full-text search cannot use: binaries, and files too large to be
// prose or source anybody reads.
func readTreeArchive(path string) (entries []index.TreeText, skipped int, err error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, 0, err
	}
	defer f.Close()
	gz, err := gzip.NewReader(f)
	if err != nil {
		return nil, 0, err
	}
	defer gz.Close()
	tr := tar.NewReader(gz)
	buf := make([]byte, treeFileMaxBytes+1)
	for {
		h, rerr := tr.Next()
		if rerr == io.EOF {
			break
		}
		if rerr != nil {
			return nil, skipped, rerr
		}
		if h.Typeflag != tar.TypeReg || h.Size > treeFileMaxBytes {
			if h.Typeflag == tar.TypeReg {
				skipped++
			}
			continue
		}
		// The archive names files under a top-level directory; the
		// index wants the repository-relative path a link uses.
		name := h.Name
		if i := strings.IndexByte(name, '/'); i >= 0 {
			name = name[i+1:]
		}
		if name == "" {
			continue
		}
		n, rerr := io.ReadFull(tr, buf[:h.Size])
		if rerr != nil && rerr != io.ErrUnexpectedEOF {
			skipped++
			continue
		}
		body := buf[:n]
		if !utf8.Valid(body) || bytes.IndexByte(body, 0) >= 0 {
			skipped++ // binary: a full-text index cannot use it
			continue
		}
		entries = append(entries, index.TreeText{
			Path: name, Body: string(body), Doc: api.IsDocPath(name),
		})
	}
	return entries, skipped, nil
}

// treeFileMaxBytes bounds one file, and treeArchiveMaxBytes the whole
// tarball. Generous for prose and source, small against a vendored
// amalgamation or a checked-in binary.
const (
	treeFileMaxBytes    = 1 << 20
	treeArchiveMaxBytes = 512 << 20
)

// Update applies a repository's settings.
func (rg *registrar) Update(name, description, visibility, homepage, topics string) error {
	if rg.odb == nil {
		return fmt.Errorf("no forge database: %w", ops.ErrUnavailable)
	}
	if err := rg.odb.SetRepoVisibility(name, visibility); err != nil {
		return fmt.Errorf("%w: %v", ops.ErrInvalid, err)
	}
	if err := rg.odb.SetRepoDescription(name, description); err != nil {
		return err
	}
	if err := rg.odb.SetRepoAbout(name, homepage, topics); err != nil {
		return err
	}
	// Visibility is cached for the read gate, so the cache moves with
	// the setting rather than a tick later. A window in which a
	// repository is still readable after being taken private is the
	// feature failing, not staleness.
	rg.refreshPrivate()
	return nil
}

// refreshPrivate reloads the private set the read gate consults.
//
// Cached because the gate runs on EVERY request and the answer changes
// only when an admin changes it. Refreshed explicitly on every write
// rather than expired on a timer: a window in which a repository is
// still readable after being taken private is not a staleness problem, it is
// the feature failing.
func (rg *registrar) refreshPrivate() {
	if rg.odb == nil {
		return
	}
	set, err := rg.odb.PrivateRepos()
	if err != nil {
		log.Printf("visibility: reload: %v", err)
		return
	}
	rg.privMu.Lock()
	rg.private = set
	rg.privMu.Unlock()
}

// CanRead reports whether the identity in ctx may read repo.
//
// Public repositories are readable by anyone, which is what keeps an
// anonymous forge working. A private one requires a role on it — and a
// site admin holds every repository, so admins are covered by the same
// lookup rather than by a special case.
func (rg *registrar) CanRead(ctx context.Context, repo string) bool {
	rg.privMu.RLock()
	private := rg.private[repo]
	rg.privMu.RUnlock()
	if !private {
		return true
	}
	user := ops.User(ctx)
	if user == "" {
		return false
	}
	if rg.odb == nil {
		return false
	}
	acct, ok := rg.odb.AccountByName(user)
	if !ok {
		return false
	}
	// RFC 0004's rule for reading at all: any read letter in the
	// effective set, categories included.
	return rg.odb.EffectiveCaps(acct.Name, repo).HasAny("gjoh2r")
}

// identityCtx resolves a sync request's Basic credentials into the
// context the visibility gate reads.
//
// The sync protocol carries its own login card, but that is verified
// deeper in (and only for pushes). For the visibility question — may
// this caller see the repository at all — the HTTP credential is what
// is available before a single byte of the body is parsed, which is
// where the decision has to be made.
func identityCtx(req *http.Request, db *oltp.DB) context.Context {
	ctx := req.Context()
	user, pass, ok := req.BasicAuth()
	if !ok || user == "" || db == nil {
		return ctx
	}
	if _, authed := db.Authenticate(user, pass); !authed {
		return ctx
	}
	return ops.WithUser(ctx, user)
}

// Delete stops serving a repository and destroys it: the store on disk,
// its derived index, and everything in the forge database keyed to it.
//
// confirm must equal the repository's name. That is the standard
// pattern because it is the only one that works: a yes/no prompt is
// answered reflexively, while typing the name requires having read
// which repository is about to go.
//
// The store on disk is removed ONLY when quarry created it — that is,
// when it lives under the repositories root. quarry serves stores it
// did not create and does not own, and the corpus this forge runs on is
// exactly that: deleting a repository named by --repo would destroy a
// checkout somebody else manages. Those refuse, and say to unregister
// instead.
func (rg *registrar) Delete(name, confirm string) error {
	if rg.odb == nil {
		return fmt.Errorf("no forge database: %w", ops.ErrUnavailable)
	}
	if confirm != name {
		return fmt.Errorf("%w: type the repository's name to confirm deletion", ops.ErrInvalid)
	}
	row, found, err := rg.odb.RepoByName(name)
	if err != nil {
		return err
	}
	if !found {
		return fmt.Errorf("no repository %q: %w", name, ops.ErrNotFound)
	}
	ours, err := rg.owns(row.Path)
	if err != nil {
		return err
	}
	if !ours {
		return fmt.Errorf("%w: %s is outside this node's repositories root, so quarry did not "+
			"create it and will not delete it — use remove to stop serving it", ops.ErrInvalid, row.Path)
	}

	// Stop serving before touching anything: Unregister waits for the
	// poll loop to exit and closes the index, and removing files under a
	// live fold is how a deletion takes the process with it.
	if err := rg.Unregister(name); err != nil && !errors.Is(err, ops.ErrNotFound) {
		return err
	}
	if err := rg.odb.PurgeRepo(name); err != nil {
		return err
	}
	rg.refreshPrivate()
	// Derived data first — it is rebuildable, so a failure here is
	// recoverable, while a half-removed store is not.
	if err := os.RemoveAll(filepath.Join(rg.dir, name)); err != nil {
		log.Printf("delete %s: derived data: %v", name, err)
	}
	return os.RemoveAll(row.Path)
}

// owns reports whether a path lies under this node's repositories root.
func (rg *registrar) owns(path string) (bool, error) {
	if rg.reposRoot == "" {
		return false, nil
	}
	root, err := filepath.Abs(rg.reposRoot)
	if err != nil {
		return false, err
	}
	abs, err := filepath.Abs(path)
	if err != nil {
		return false, err
	}
	return strings.HasPrefix(abs, root+string(os.PathSeparator)), nil
}

// Deletable reports whether this node created the repository's store.
func (rg *registrar) Deletable(name string) bool {
	if rg.odb == nil {
		return false
	}
	row, found, err := rg.odb.RepoByName(name)
	if err != nil || !found {
		return false
	}
	ours, err := rg.owns(row.Path)
	return err == nil && ours
}