Artifact [48c4f01bb2]

Artifact 48c4f01bb2f2065d854feaa8c296447534491518d89d62ea347124aa0d5a26eb:

• File cmd/quarry/registrar.go — part of check-in [6ddc78b549] at 2026-10-03 00:48:35 on branch trunk — land admins-send-reset-links

  Checks tested the merge with trunk at 83aec99ff875; trunk has moved since. (user: dmestas, size: 87024) [annotate] [blame] [check-ins using] [more...]

package main

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

	"craftdesign.group/quarry/internal/api"
	"craftdesign.group/quarry/internal/ci"
	"craftdesign.group/quarry/internal/errlog"
	"craftdesign.group/quarry/internal/events"
	"craftdesign.group/quarry/internal/index"
	"craftdesign.group/quarry/internal/ingest"
	"craftdesign.group/quarry/internal/lease"
	"craftdesign.group/quarry/internal/obs"
	"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"

	"craftdesign.group/quarry/internal/fault"
)

// 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 {
	// schedules is each repository's scheduled workflows as of its
	// default tip, shared by the scheduler and the CI page.
	schedules   *ci.ScheduleCache
	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
	// drainRepo, when set, stands in for the registered repository's
	// read side in drainDeferred.
	drainRepo func(name string) (ci.Repo, bool)

	ev         *events.Server
	leases     *lease.Manager
	jobs       *jobGate
	deps       *api.Deps
	fronter    *xfer.Fronter
	ingestLane *work.Lane
	bulk       *work.Lane
	engine     *ci.Engine
	odb        *oltp.DB
	// doors resolves tokens at the sync lane's Basic round and counts
	// the misses; nil resolves them uncounted.
	doors *tokenDoors
	// openLandings is, per repository, the open landings the fold loop
	// has already seen (id -> branch), so the ones a fold ADDS are the
	// ones that just opened and the ones it DROPS the ones that just
	// closed. Seeded on the first sweep without asking for anything: a
	// restart is not a thousand landings opening at once.
	landingsMu   sync.Mutex
	openLandings map[string]map[string]string
	// pipelineless remembers branch tips that have no pipelines, so the
	// pair sweep does not list their files again on every fold.
	pipelineless sync.Map
	// gateSeen is when the gate sweep first saw an unaccepted check-in
	// on a protected branch (key repo NUL hash). A forge accept records
	// itself a moment AFTER its merge is folded, so a sweep in that gap
	// read the forge's own merge as a bypass; a finding now waits
	// gateGrace from first sight — the forge's clock, not the check-in's
	// author-written date, so a future-dated push cannot wait forever.
	gateSeen    sync.Map
	gateRecheck sync.Map // repo -> struct{}: a re-sweep is scheduled
	gateGrace   time.Duration
	clock       func() time.Time
	// ciTagsDefault is the fleet's verdict-tag default (RFC 0009); a
	// repository's own mode overrides it either way.
	ciTagsDefault bool
	// carried holds the merges carryForward wrote a verdict onto that
	// the index does not show yet, keyed by repository and merge hash:
	// a fork and its source share hashes, and each needs its own.
	carried sync.Map
	// repoQuota caps how many repositories a non-site-admin account may
	// administer at once (--repo-quota, default 25). Creation is open
	// to any account now, and an open door needs a limit; a site admin
	// is never subject to it.
	repoQuota int

	// 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

	// private is the visibility list the read gate consults, loaded at
	// privAt. privGen numbers loads as they begin and privStored is the
	// number of the one in private, so a slow load cannot replace a
	// newer one. privBusy is set while a background reload runs.
	// privStored is zero until a load has succeeded: there is no list
	// to serve yet, so everything answers private (privateSet).
	privMu     sync.RWMutex
	private    map[string]bool
	privAt     time.Time
	privGen    uint64
	privStored uint64
	privBusy   atomic.Bool
	privLoad   func() (map[string]bool, error) // nil: odb.PrivateRepos
}

// privateTTL bounds how long another node's visibility change can go
// unseen here: every node shares the OLTP but caches the list.
const privateTTL = 5 * time.Second

// 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.
// leafAnswer decides whether a cached leaf set may be served. It is a
// function so the rule can be tested: the closure below cannot be, and
// this rule is the difference between a mark that states a fact and one
// that states a fact that WAS true.
//
// A set derived before the DAG moved is not "slightly out of date" — a
// merge is precisely what retires a leaf, so the stale set's marks are
// wrong exactly where the reader is looking.
func leafAnswer(set map[string]bool, cachedGen, gen uint64) (map[string]bool, bool) {
	if set != nil && cachedGen == gen {
		return set, true
	}
	return nil, false
}

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(lease.Token) error) error {
			return leaseRefusal(rg.leases.Guard(code, fn))
		},
	}
	// 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 fresh, ok := leafAnswer(leafSet, leafGen, gen); ok {
			return fresh, leafErr
		}
		if !leafWarming {
			leafWarming = true
			go warmLeaves(gen)
		}
		// The DAG has moved since this set was derived, so nil rather
		// than the old set — even though the old set is "almost right".
		//
		// A leaf mark is not decoration, it is a claim: this check-in is
		// the tip of its branch and nothing follows it. The instant that
		// stops being true is a merge, and a merge is exactly when
		// somebody is looking at the timeline. Serving the previous set
		// then puts a ring on the two check-ins that just STOPPED being
		// leaves — the reader is told the branch is still open at the
		// moment it closed, which is the one answer the mark must never
		// give.
		//
		// Missing marks are the honest failure and the one this file
		// already accepts while cold: no ring says "not known", a wrong
		// ring says "still open". The warm above is already running, so
		// the gap lasts one derivation.
		return nil, 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)
	// Warmed in the background for the same reason as the leaf set: a
	// subprocess on a render path is latency one reader pays for
	// everybody, and one running on a detached context is invisible in
	// the footer while still counting toward the page's total. With
	// both warmed, "ops minus zit" means what the ops panel says it
	// means — index and OLTP work — instead of hiding a subprocess.
	var dbWarming bool
	warmDefault := func(gen uint64) {
		tip, name, err := cli.Info(context.Background(), "trunk")
		dbMu.Lock()
		if err == nil {
			dbName, dbTip, dbGen = name, tip, gen
		}
		dbWarming = false
		dbMu.Unlock()
	}
	entry.DefaultBranch = func() (string, string) {
		gen := ingestGen.Load()
		dbMu.Lock()
		defer dbMu.Unlock()
		if dbGen == gen {
			return dbName, dbTip
		}
		if !dbWarming {
			dbWarming = true
			go warmDefault(gen)
		}
		// Empty means the caller falls back to the index heuristic,
		// which is what it already did on error.
		return dbName, dbTip
	}
	go func() {
		dbMu.Lock()
		if dbWarming {
			dbMu.Unlock()
			return
		}
		dbWarming = true
		gen := ingestGen.Load()
		dbMu.Unlock()
		warmDefault(gen)
	}()
	entry.Mirror = func(url string) (bool, error) {
		var changed bool
		err := entry.Guard(func(tok lease.Token) error {
			var perr error
			changed, perr = cli.Pull(tok, url)
			return perr
		})
		return changed, err
	}
	pipe := ingest.NewPipeline(cli, rg.ev, ix, name, rg.node)
	// In-process reindex: a rebuild, so it runs on one of the pipeline's
	// rebuild slots and a fleet-wide repair is staged maxRebuilds stores
	// at a time rather than a stampede. It does NOT take an ingest-lane
	// slot: it waits for the store's lock, and nothing may wait for that
	// holding a lane slot (see ingest.Admit). 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) {
		indexing.Store(true)
		defer indexing.Store(false)
		n, err := pipe.Reindex()
		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
	// syncIn is the round the poll runs, admitted through the ingest lane;
	// entry.Sync is the same round for
	// the write ops, which hold no lane slot.
	syncIn := func(admit ingest.Admit) (int, error) {
		n, err := pipe.SyncIn(admit)
		// 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.Sync = func() (int, error) { return syncIn(nil) }
	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)
	// The read gate's cache moves with the repository, not with a tick
	// after it.
	//
	// refreshPrivate has always been called by the writes that change a
	// repository's VISIBILITY — Update and Delete — and once at the end
	// of startup. Creation was the gap: a repository that arrived
	// PRIVATE at runtime was served, and missing from the set the gate
	// consults, until a restart or an unrelated edit refreshed it.
	//
	// The window is not cosmetic. IsPrivate answers false for it, so the
	// canonical redirect announces the namespace to a stranger — "it is
	// at /r/<owner>/<name>" — which is precisely the sentence the gate
	// exists to withhold, and the sync gate reads the same set. The
	// guard was right and the cache was stale, which is why the code
	// looked correct while the hole was open.
	//
	// Here rather than at each call site because Register is the one
	// place a repository starts being served: startup, repo.add and the
	// clone importer all arrive through it, so the property holds by
	// construction instead of by remembering.
	rg.refreshPrivate()

	ctx, cancel := context.WithCancel(context.Background())
	// Stopping the pipeline too: a round queued for a rebuild slot does
	// not watch the poller's context, and Unregister waits for the poller
	// to exit.
	stop := func() {
		cancel()
		pipe.Stop()
	}
	done := make(chan struct{})
	rg.track(name, &served{ix: ix, stop: stop, done: done})
	go rg.poll(ctx, done, name, path, cli, ix, syncIn, 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's round taking an admission (syncIn) — 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(ingest.Admit) (int, error), indexing *atomic.Bool) {
	defer close(done)

	start := time.Now()
	n, err := sync(func(body func(leave func()) error) error {
		return rg.ingestLane.DoNamedLeavable(ctx, "first ingest", name, body)
	})
	indexing.Store(false)
	switch {
	case errors.Is(err, ingest.ErrStopped):
		// Removed or shutting down while queued for a rebuild: not a failure.
	case errors.Is(err, work.ErrBusy):
		// Back-pressure at boot, not a failure: the round loop retries,
		// and journals only a store refused past ingestShedDeadline.
		log.Printf("initial ingest of %s deferred: the ingest lane is full", name)
	case err != nil:
		log.Printf("initial ingest of %s: %v", name, err)
	default:
		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()
	// A round costs a zit subprocess and one of --ingest-limit lane
	// slots, and a store nobody has written since the last round has
	// nothing to ingest. With nineteen stores ticking every two seconds
	// through a three-wide lane, unconditional rounds kept the lane
	// saturated and shed the polls that mattered — quarry's own
	// check-ins went unseen and CI never fired. A stat is free, so the
	// tick reads the store's stamp and only enters the lane when it
	// moved. The stamp is taken BEFORE the round: a write that lands
	// during one is picked up by the next.
	var ingested storeStamp
	var shed shedBackoff
	for {
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
		}
		stamp := stampStore(path)
		if stamp.ok && stamp == ingested {
			continue
		}
		ran, err := rg.ingestRound(ctx, time.Now(), name, &shed, func(admit ingest.Admit) error {
			_, serr := sync(admit)
			return serr
		})
		if !ran {
			continue // shed or backing off: back-pressure, handled there
		}
		if 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)
				errlog.Default.ReportIn("ingest", "ingest "+name, name, "",
					fault.Wrap(err, "a store could not be ingested").
						Cost("this store's new check-ins are not searchable or listed; the store itself is intact").
						With("repo", name))
			}
			continue
		}
		ingested = stamp
		// 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)
					warmCtx := obs.WithReason(context.Background(), obs.ReasonWarm)
					if _, err := cli.Ls(warmCtx, tip); err == nil {
						warmedTip.Store(tip)
					}
					warmFileHistory(cli, tip, name)
				}()
			}
		}
		rg.afterFold(name, cli, ix)
	}
}

// afterFold runs the store's jobs that react to a fold, on the node
// holding its lease only. The fold itself runs on every node: each
// keeps its own index.
func (rg *registrar) afterFold(name string, cli *zitcli.Client, ix *index.Index) {
	if !rg.jobs.store(name) {
		return
	}
	// CI looks right after a fold, which is the moment a new
	// check-in becomes visible to quarry whatever brought it in.
	// A companion store holds no check-ins and has no pipeline
	// (RFC 0015): nothing to run, and nothing to wake the engine.
	if rg.engine.Enabled() && !rg.isCompanion(name) {
		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)
		}
		// A tag added since the last fold (RFC 0020's tag event): after
		// the check-ins, so a check-in that arrives tagged runs its push
		// workflows first.
		if err := ci.TriggerTags(rg.engine, ciStore{db: rg.odb}, name,
			ciRepo{ix: ix, zit: cli}, rg.ciCloneBase+name); err != nil {
			log.Printf("ci: %s tags: %v", name, err)
		}
	}
	rg.gateSweep(name, ix)
	rg.carryForward(name, ix)
	rg.landingSweep(name, ix)
	rg.closeSweep(name, ix)
}

// drainDeferred releases check-ins waiting for a worker slot on the
// stores this node runs jobs for. The fold's ci.Trigger drains only
// when a store changes, so on a quiet repository a wait outlived the
// room that ended it (quarry#296); this runs on a timer instead. Which
// repositories wait is one query on the OLTP table, so a tick costs
// nothing for the ones that have none.
func (rg *registrar) drainDeferred() {
	if rg.odb == nil || !rg.engine.Enabled() {
		return
	}
	repos, err := rg.odb.CIDeferredRepos()
	if err != nil {
		log.Printf("ci: deferred list: %v", err)
		return
	}
	for _, name := range repos {
		if !rg.jobs.store(name) || rg.isCompanion(name) {
			continue
		}
		var cr ci.Repo
		if rg.drainRepo != nil {
			var ok bool
			if cr, ok = rg.drainRepo(name); !ok {
				continue
			}
		} else {
			repo, ok := rg.deps.Repos.Get(name)
			if !ok || repo.Zit == nil || repo.Index == nil {
				continue
			}
			cr = ciRepo{ix: repo.Index, zit: repo.Zit}
		}
		ci.Drain(rg.engine, ciStore{db: rg.odb}, name, cr, rg.ciCloneBase+name)
	}
}

// landingSweep runs after every fold. It keeps every open landing's
// merge tested (pairSweep), and tells the accepters of each landing this
// fold OPENED that it is waiting on them: a landing opened by push,
// which is every one zit opens, never passes through landing.open.
func (rg *registrar) landingSweep(name string, ix *index.Index) {
	if rg.deps == nil || ix == nil {
		return
	}
	open, err := ix.Landings("open", 500)
	if err != nil {
		return
	}
	rg.landingsMu.Lock()
	if rg.openLandings == nil {
		rg.openLandings = map[string]map[string]string{}
	}
	seen, seeded := rg.openLandings[name]
	if !seeded {
		seen = map[string]string{}
		for _, l := range open {
			seen[l.ID] = l.Branch
		}
		rg.openLandings[name] = seen
		rg.landingsMu.Unlock()
		rg.pairSweep(name, ix, open, nil)
		return
	}
	var fresh []index.LandingSummary
	opening := map[string]bool{}
	stillOpen := map[string]bool{}
	openBranch := map[string]bool{}
	for _, l := range open {
		stillOpen[l.ID], openBranch[l.Branch] = true, true
		if _, ok := seen[l.ID]; !ok {
			seen[l.ID] = l.Branch
			fresh = append(fresh, l)
			opening[l.ID] = true
		}
	}
	var closed []string
	for id, branch := range seen {
		if stillOpen[id] {
			continue
		}
		delete(seen, id)
		// A landing superseded by a newer one on the same branch is
		// closed, but its branch's merge is the newer one's to finish.
		if !openBranch[branch] {
			closed = append(closed, branch)
		}
	}
	rg.landingsMu.Unlock()
	rg.pairSweep(name, ix, open, opening)
	for _, l := range fresh {
		rg.deps.NotifyOpenedLanding(name, l)
	}
	for _, branch := range closed {
		rg.cancelClosedPairs(name, ix, branch)
	}
}

// cancelClosedPairs stops a closed landing's merge runs that are still
// going. Accepted or rejected, the landing is decided: a verdict on its
// merge answers nobody, and the runs hold runner slots that the check-in
// the accept wrote is waiting for. Only the runs the landing sweep
// started are the landing's (subject pair); a merge a push ran is the
// push's (push-pair, RFC 0009 §once), its tip's verdict, and stays, as
// does a run whose labels the engine cannot show. Best-effort: a run
// that finishes in the race is simply done.
//
// If the tip is then left with nothing that passed, failed or is still
// going, it is tested as itself: a branch is never left with no verdict.
func (rg *registrar) cancelClosedPairs(name string, ix *index.Index, branch string) {
	if !rg.engine.Enabled() || rg.odb == nil {
		return
	}
	tip, err := ix.BranchTip(branch)
	if err != nil || tip == "" {
		return
	}
	runs, err := rg.odb.UnfinishedCIRuns()
	if err != nil {
		return
	}
	prefix := "pair:" + tip + ":"
	cancelled := map[string]bool{}
	for _, r := range runs {
		if r.Repo != name || !strings.HasPrefix(r.Pair, prefix) {
			continue
		}
		if run, err := rg.engine.Status(r.RunID); err != nil || run.Labels["subject"] != "pair" {
			continue
		}
		if err := rg.engine.Cancel(r.RunID); err != nil {
			log.Printf("ci: %s: cancelling %s's merge run %s: %v", name, branch, r.RunID, err)
			continue
		}
		cancelled[r.RunID] = true
		log.Printf("ci: %s: %s's landing closed; cancelled its merge run %s", name, branch, r.RunID)
	}
	if rg.tipAnswered(name, tip, cancelled) || rg.deps == nil {
		return
	}
	repo, ok := rg.deps.Repos.Get(name)
	if !ok || repo.Zit == nil {
		return
	}
	if err := ci.RunTip(rg.engine, ciStore{db: rg.odb}, name, ciRepo{ix: ix, zit: repo.Zit},
		rg.ciCloneBase+name, ci.Checkin{Hash: tip, Branch: branch}); err != nil {
		log.Printf("ci: %s: testing %s's tip %.12s after its landing closed: %v", name, branch, tip, err)
	}
}

// tipAnswered reports whether a branch tip has a verdict, or one on its
// way: a run of the tip or of a merge of it that passed, failed or is
// still going, other than the runs just cancelled.
func (rg *registrar) tipAnswered(name, tip string, cancelled map[string]bool) bool {
	own, err := rg.odb.CIRunsFor(name, tip)
	if err != nil {
		return true // cannot tell: starting more is the worse mistake
	}
	merges, err := rg.odb.CIRunsForTip(name, tip)
	if err != nil {
		return true
	}
	for _, r := range append(own, merges...) {
		if cancelled[r.RunID] {
			continue
		}
		switch {
		case (ci.Run{Status: r.Status}).Passed(), r.Status == "failed", !ciTerminalStatus(r.Status) && r.Status != "deferred":
			return true
		}
	}
	return false
}

// pairSweep keeps each open landing tested as the merge with its
// target, as a pull request's checks run on its test merge (quarry#290):
// a branch tip with no pair yet is merged into the target as it is and
// tested. A target that moves re-runs nothing, as it re-runs nothing on
// a pull request; the landing says which target tip its checks tested,
// and the next push to the branch tests the merge with the target as
// it is then.
//
// A landing opening (opening holds its id) over a tip a push already
// tested as a merge (RFC 0009 §once) runs what that merge has no
// verdict for, when the target tip is the same: the landing-only
// pipelines, and any run that errored or was cancelled. Against a
// target that has moved since, the merge is a new one and runs whole.
func (rg *registrar) pairSweep(name string, ix *index.Index, open []index.LandingSummary, opening map[string]bool) {
	if !rg.engine.Enabled() || rg.odb == nil || rg.deps == nil {
		return
	}
	repo, ok := rg.deps.Repos.Get(name)
	if !ok || repo.Zit == nil {
		return
	}
	st, cr := ciStore{db: rg.odb}, ciRepo{ix: ix, zit: repo.Zit}
	for _, l := range open {
		if l.SupersededBy != "" {
			continue
		}
		tip, err := ix.BranchTip(l.Branch)
		if err != nil || tip == "" {
			continue
		}
		if _, none := rg.pipelineless.Load(name + "\x00" + tip); none {
			continue
		}
		// Any merge of the tip, cancelled ones included: a merge somebody
		// cancelled is not started again on every sweep.
		if tried, err := rg.odb.CIRunsForTip(name, tip); err != nil || (len(tried) > 0 && !opening[l.ID]) {
			continue
		}
		target, err := ix.BranchTip(l.Target)
		if err != nil || target == "" {
			continue
		}
		ran, err := ci.RunPairAt(rg.engine, st, name, cr, rg.ciCloneBase+name, l.Branch, tip, target)
		if err != nil {
			log.Printf("ci: %s: landing %.12s: merge of %.12s into %.12s: %v", name, l.ID, tip, target, err)
			continue
		}
		if !ran {
			rg.pipelineless.Store(name+"\x00"+tip, struct{}{})
		}
	}
}

// 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
		}
		held := false
		for _, e := range fresh {
			if e.Branch != rule.Branch {
				continue
			}
			key := name + "\x00" + e.Hash
			if ok, _ := rg.odb.ForgeAccepted(name, e.Hash); ok {
				rg.gateSeen.Delete(key)
				continue
			}
			first, _ := rg.gateSeen.LoadOrStore(key, rg.now())
			if rg.now().Sub(first.(time.Time)) < rg.grace() {
				held = true
				continue
			}
			rg.gateSeen.Delete(key)
			if err := rg.odb.AddGateBypass(name, rule.Branch, e.Hash, e.User); err == nil {
				log.Printf("gate: %s %s moved past its checks at %.12s by %s — no forge accept behind it", name, rule.Branch, e.Hash, e.User)
			}
		}
		if held {
			// Leave the watch where it is so the held check-in is looked
			// at again, and look again once the grace is up even if
			// nothing else is written to this store meanwhile.
			if _, busy := rg.gateRecheck.LoadOrStore(name, struct{}{}); !busy {
				time.AfterFunc(rg.grace()+time.Second, func() {
					rg.gateRecheck.Delete(name)
					if rg.jobs.store(name) {
						rg.gateSweep(name, ix)
					}
				})
			}
			continue
		}
		_ = rg.odb.SetGateWatch(name, rule.Branch, b.Tip)
	}
}

func (rg *registrar) now() time.Time {
	if rg.clock != nil {
		return rg.clock()
	}
	return time.Now()
}

func (rg *registrar) grace() time.Duration {
	if rg.gateGrace > 0 {
		return rg.gateGrace
	}
	return time.Minute
}

// 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
		}
		key := name + "\x00" + m.Checkin
		if v, verr := ix.VerdictOf(m.Checkin); verr != nil || v != "" {
			if v != "" {
				rg.carried.Delete(key)
			}
			continue
		}
		// Written on an earlier pass whose ingest failed: the tag is in
		// the store and the index has not caught up. The tag log is
		// append-only, so writing again would mint a second verdict.
		if _, written := rg.carried.Load(key); written {
			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
		}
		// After a restart nothing is remembered and the index may still
		// be behind: the store is asked before a second verdict is minted.
		if repo.Zit != nil {
			if tags, terr := repo.Zit.ActiveTags(context.Background(), m.Checkin); terr == nil {
				_, pass := tags["ci-pass"]
				_, fail := tags["ci-fail"]
				if pass || fail {
					rg.carried.Store(key, struct{}{})
					continue
				}
			}
		}
		err := repo.WriteTag(verdict, m.Checkin, "premerge")
		if err != nil && !errors.Is(err, api.ErrTagNotIndexed) {
			log.Printf("ci: carry-forward %s %.12s: %v", name, m.Checkin, err)
			continue
		}
		rg.carried.Store(key, struct{}{})
		if err != nil {
			log.Printf("ci: %s %.12s %s (carried from premerge) written, index lags: %v", name, m.Checkin, verdict, 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; by is
// who asked, the runs' starter.
func (rg *registrar) TriggerCI(name, checkin, by string) error {
	if !rg.engine.Enabled() {
		return fmt.Errorf("no CI engine configured")
	}
	if rg.isCompanion(name) {
		return fmt.Errorf("%s is a forum store and has no CI", name)
	}
	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, by)
}

// TriggerPairCI re-runs every pipeline on a landing's merge
// (ci.RerunPairAt): the Re-run on a landing and on a merge run's page.
func (rg *registrar) TriggerPairCI(name, branch, branchTip, targetTip 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.RerunPairAt(rg.engine, ciStore{db: rg.odb}, name,
		ciRepo{ix: repo.Index, zit: repo.Zit}, rg.ciCloneBase+name, branch, branchTip, targetTip)
}

// FireSchedules launches every repository's scheduled workflows due in
// minutes (quarry owns schedules: RFC 0019, RFC 0021 §runs). lead is
// the fleet lease, asked again before every claim. Skipped:
// companions, repositories whose admin said no to CI, and repositories
// no runner serves.
func (rg *registrar) FireSchedules(minutes []time.Time, lead func() bool) {
	if !rg.engine.Enabled() || rg.odb == nil || len(minutes) == 0 {
		return
	}
	workers, werr := rg.engine.Workers()
	names := rg.deps.Repos.Names()
	rg.schedules.Retain(names)
	for _, name := range names {
		if !lead() {
			return
		}
		repo, ok := rg.deps.Repos.Get(name)
		if !ok || repo.Index == nil || repo.Zit == nil || rg.isCompanion(name) {
			continue
		}
		if who, _ := rg.odb.CIDeclined(name); who != "" {
			continue
		}
		if werr == nil && !ci.RunnerFor(workers, name, time.Now()) {
			continue
		}
		if err := ci.FireSchedules(rg.engine, ciStore{db: rg.odb}, rg.schedules, name,
			ciRepo{ix: repo.Index, zit: repo.Zit}, rg.ciCloneBase+name, minutes, lead); err != nil {
			log.Printf("ci: %s schedules: %v", name, err)
		}
	}
}

// CISchedules lists a repository's scheduled workflows from the cache
// the scheduler reads, each with its next fire or why it does not, and
// its last failed minute.
func (rg *registrar) CISchedules(name string) ([]ci.Scheduled, error) {
	repo, ok := rg.deps.Repos.Get(name)
	if !ok || repo.Index == nil || repo.Zit == nil {
		return nil, fmt.Errorf("no repository %q", name)
	}
	out, err := rg.schedules.Schedules(name, ciRepo{ix: repo.Index, zit: repo.Zit}, time.Now())
	if err != nil || rg.odb == nil {
		return out, err
	}
	for i := range out {
		out[i].LastFailure, out[i].LastFailureAt = rg.odb.LastScheduleFailure(name, out[i].Pipeline)
	}
	return out, nil
}

// isCompanion answers whether a store is some repository's forum
// (RFC 0015). Without an OLTP nothing is a companion.
func (rg *registrar) isCompanion(name string) bool {
	if rg.odb == nil {
		return false
	}
	_, _, ok, err := rg.odb.CompanionParent(name)
	return err == nil && ok
}

// 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 oltp.IsReserved(spec.Name) {
		return "", fmt.Errorf("%w: %q is a name the forge keeps for its own routes", ops.ErrInvalid, spec.Name)
	}
	// The third direction of the name-pool rule, and the one that lives
	// here because this is where repositories are made.
	//
	// A repository path's FIRST segment has to be readable without
	// ambiguity: /r/A/B is either the repository A with B beneath it,
	// or the repository B under the namespace A, and nothing may make
	// both true. oltp refuses an account or team named after a
	// repository; this refuses the reverse. With all three the pools
	// are disjoint and the reading is arithmetic rather than a guess.
	if rg.odb != nil {
		if _, taken := rg.odb.AccountByName(spec.Name); taken {
			return "", fmt.Errorf("%w: %q is an account's name, and a repository cannot take it: "+
				"an account's name is where its repositories appear", ops.ErrInvalid, spec.Name)
		}
		if rg.odb.TeamNameTaken(spec.Name) {
			return "", fmt.Errorf("%w: %q is a team's name, and a repository cannot take it: "+
				"a team's name is where its repositories appear", ops.ErrInvalid, spec.Name)
		}
	}
	// The quota is per ACCOUNT, not per request: a site admin runs the
	// forge's capacity and is never limited, so this only ever refuses
	// someone who has hit their own ceiling.
	if spec.Author != "" {
		if acct, ok := rg.odb.AccountByName(spec.Author); !ok || !acct.Admin {
			n, err := rg.odb.OwnedRepoCount(spec.Author)
			if err != nil {
				return "", err
			}
			if n >= rg.repoQuota {
				return "", fmt.Errorf("%w: %s already owns %d repositories, "+
					"the most this forge allows", ops.ErrInvalid, spec.Author, rg.repoQuota)
			}
		}
	}
	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)
		}
		rg.claim(spec)
		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)
	}
	rg.claim(spec)
	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.DoNamed(context.Background(), "clone", spec.Name, 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)
		}
		// The mirror was recorded when the import was accepted, on the
		// expectation of a store to pull into. There is none: without
		// this the scheduler asked for a repository that does not exist
		// every tick and journalled an error each time — two thousand
		// of them for one refused clone (libspatialite, 2026-09-06).
		if spec.Mirror {
			if derr := rg.odb.DeleteMirror(spec.Name); derr != nil {
				log.Printf("import %s: forgetting mirror: %v", spec.Name, derr)
			}
		}
		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 }

// api.AccountAdmin is reached through a runtime type assertion, so a
// method missed below does not fail the build — it silently turns off
// invites, resets, the lockout and the session audit TOGETHER, and the
// first anybody hears of it is a surface reporting that this forge has
// no invites. Assert it here, where the compiler can say so instead.
var _ api.AccountAdmin = accountStore{}

// The private-beta half (api.AccountAdmin).
func (s accountStore) InviteAccount(name, email string, admin bool) (string, error) {
	return s.db.InviteAccount(name, email, admin)
}
func (s accountStore) InviteFor(token string) (string, bool) { return s.db.InviteFor(token) }
func (s accountStore) RedeemInvite(token, password string) (string, error) {
	return s.db.RedeemInvite(token, password)
}
func (s accountStore) ResetPassword(name string) (string, error) {
	return s.db.ResetPassword(name)
}
func (s accountStore) TokenKind(token string) string          { return s.db.TokenKind(token) }
func (s accountStore) SetDisabled(name string, on bool) error { return s.db.SetDisabled(name, on) }
func (s accountStore) Sessions(name string) ([]api.SessionInfo, error) {
	rows, err := s.db.Sessions(name)
	if err != nil {
		return nil, err
	}
	out := make([]api.SessionInfo, 0, len(rows))
	for _, x := range rows {
		out = append(out, api.SessionInfo{ID: x.ID, Account: x.Account, Created: x.Created, Expires: x.Expires,
			LastSeen: x.LastSeen, IP: x.IP, Agent: x.Agent})
	}
	return out, nil
}
func (s accountStore) EndSession(id, account string) error { return s.db.EndSessionID(id, account) }
func (s accountStore) EndSessions(account string) error    { return s.db.EndAccountSessions(account) }

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,
			Email: r.Email, EmailPending: r.EmailPending,
			Disabled: r.Disabled, Pending: r.Pending, LastLogin: r.LastLogin})
	}
	return out, nil
}

// The confirmed-address half (api.AccountEmail).
func (a accountStore) AccountByName(name string) (api.Account, bool) {
	r, ok := a.db.AccountByName(name)
	if !ok {
		return api.Account{}, false
	}
	return api.Account{Name: r.Name, Admin: r.Admin, Created: r.Created,
		Email: r.Email, EmailPending: r.EmailPending,
		Disabled: r.Disabled, Pending: r.Pending, LastLogin: r.LastLogin}, true
}
func (a accountStore) SetEmail(name, email string) error { return a.db.SetEmail(name, email) }
func (a accountStore) SetPendingEmail(name, address string) (string, error) {
	return a.db.SetPendingEmail(name, address)
}
func (a accountStore) ConfirmEmail(account, code string) (string, bool, error) {
	return a.db.ConfirmEmail(account, code)
}
func (a accountStore) ClearEmail(name string) error { return a.db.ClearEmail(name) }

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

func (a accountStore) SetOperator(name, operator string) error {
	return a.db.SetOperator(name, operator)
}

// VerifyPassword and SetPasswordSelf are api.AccountAuth: the
// self-service half of me.passwd. No forge op sets another account's
// password; that is `quarry account passwd`, on the box (#319).
func (a accountStore) VerifyPassword(name, password string) bool {
	return a.db.VerifyPassword(name, password)
}
func (a accountStore) SetPasswordSelf(name, password, keep string) error {
	return a.db.SetPasswordSelf(name, password, keep)
}

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

// Grant and Revoke are the API's writes, each judged by the RFC 0008
// guard for the actor under the lock it writes with. A site admin
// removing a repository's last admin is warned, not refused; anyone
// else is refused.
func (a accountStore) Grant(actor, account, repo, role string) (string, error) {
	r, err := oltp.ParseRole(role)
	if err != nil {
		return "", err
	}
	return a.db.GrantGuarded(actor, account, repo, r)
}

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

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()
	}
	return out, nil
}

// 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.
//
// Four rather than two since zit 0.3.2, which cut `leaves` on sqlite
// from 1.15s to 0.26s. The warm pass is a race against the first
// reader — until a repository is warmed, its About panel pays ~3 zit
// cats on the render path, which measured 543-775ms — so finishing the
// sweep sooner is what shortens that window. Still bounded well below
// the store count, because the point of the cap is to not spike memory
// on the biggest repositories with nobody waiting.
var warmSem = make(chan struct{}, 4)

// 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 }()
	// The home page's activity chart: a scan of the whole history, 4.8s
	// on tcl. After the first read a fold only marks it stale and it is
	// refreshed in the background, but a fresh process has nothing to
	// serve, so the first reader after every deploy paid the scan. It
	// is built here, at boot, under the same four-at-a-time bound.
	_, _ = ix.ActivityDays("")
	// The dashboard's kind chips: a count of every event, 0.46s on
	// pkgsrc with a cold page cache. The dashboard asks every
	// repository, so after a deploy its first load paid them all.
	// Deferred so it runs after the zit reads below, which the first
	// reader of this store waits on, and on every return path.
	defer func() { _, _ = ix.EventKinds() }()
	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
)

func (rg *registrar) Update(name string, edit api.RepoEdit, by string) error {
	if rg.odb == nil {
		return fmt.Errorf("no forge database: %w", ops.ErrUnavailable)
	}
	// One transactional write for every setting, so the change and the
	// record of who made it cannot come apart, and the before-value in
	// that record is read where nothing can move it. The rule about
	// mirrorable during a private detour lives in there too: it needs
	// the authoritative visibility, which is the row the transaction is
	// already holding.
	if _, err := rg.odb.PatchRepo(name, oltp.RepoPatch{
		Description: edit.Description,
		Visibility:  edit.Visibility,
		Homepage:    edit.Homepage,
		Topics:      edit.Topics,
		Chat:        edit.Chat,
		Mirrorable:  edit.Mirrorable,
		UV:          edit.UV,
	}, by); err != nil {
		return fmt.Errorf("%w: %v", ops.ErrInvalid, 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, and
// returns once the new set is in place.
//
// Cached because the gate runs on EVERY request. This node's own
// visibility writes call it directly, because a window in which a
// repository is still readable after being taken private is the feature
// failing, not staleness. Another node's writes reach the shared OLTP
// only, so reads also reload a list older than privateTTL (privateSet).
//
// A failed load keeps the previous set and counts as a load for the
// TTL, so a broken database is retried every privateTTL, not per read.
func (rg *registrar) refreshPrivate() {
	load := rg.privLoader()
	if load == nil {
		return
	}
	rg.privMu.Lock()
	rg.privGen++
	gen := rg.privGen
	rg.privMu.Unlock()

	set, err := load()

	rg.privMu.Lock()
	if gen > rg.privStored {
		rg.privAt = rg.now()
		if err == nil {
			rg.private, rg.privStored = set, gen
		}
	}
	rg.privMu.Unlock()
	if err != nil {
		log.Printf("visibility: reload: %v", err)
	}
}

func (rg *registrar) privLoader() func() (map[string]bool, error) {
	if rg.privLoad != nil {
		return rg.privLoad
	}
	if rg.odb == nil {
		return nil
	}
	return rg.odb.PrivateRepos
}

// privateSet returns the cached set, and whether one has ever loaded.
// One older than privateTTL starts a single background reload and is
// still served meanwhile: a request never waits on the reload, and
// never sees an empty set because one is running. The set is replaced
// whole, never mutated, so it is safe to read after the lock is
// released.
func (rg *registrar) privateSet() (set map[string]bool, loaded bool) {
	load := rg.privLoader()
	if load == nil {
		return nil, true // no forge database: nothing can be private
	}
	rg.privMu.RLock()
	set, at, loaded := rg.private, rg.privAt, rg.privStored > 0
	rg.privMu.RUnlock()
	if rg.privStale(at) && rg.privBusy.CompareAndSwap(false, true) {
		// Checked again now the flag is held: a reload that finished
		// between the read above and the swap has already stamped
		// privAt, and must not be followed by another.
		rg.privMu.RLock()
		at = rg.privAt
		rg.privMu.RUnlock()
		if !rg.privStale(at) {
			rg.privBusy.Store(false)
			return set, loaded
		}
		go func() {
			defer rg.privBusy.Store(false)
			rg.refreshPrivate()
		}()
	}
	return set, loaded
}

func (rg *registrar) privStale(at time.Time) bool {
	return rg.now().Sub(at) > privateTTL
}

// IsPrivate answers from the cached set refreshPrivate maintains. It is
// the first half of CanRead, exposed because the web layer's canonical
// redirect has to ask the question before it has resolved an identity.
//
// Before any load has succeeded — the boot load failed — it answers
// true for every repository: the forge is hidden, not opened.
func (rg *registrar) IsPrivate(repo string) bool {
	set, loaded := rg.privateSet()
	return !loaded || set[repo]
}

// 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, from a
// grant or a team. A site admin is no exception (#318): the flag is s
// at forge scope, never a read letter on somebody's repository.
func (rg *registrar) CanRead(ctx context.Context, repo string) bool {
	if !rg.IsPrivate(repo) {
		return true
	}
	user := ops.User(ctx)
	if user == "" {
		return false
	}
	if rg.odb == nil {
		return false
	}
	// RFC 0004's rule for reading at all: any read letter in the
	// effective set, categories included. A token's set when the caller
	// came in on one, asked first: the resolver already stood behind
	// its account, and a run token (RFC 0021) has none.
	if tok, ok := ops.TokenOf(ctx); ok {
		return tok.Caps(repo).HasAny("gjoh2r")
	}
	acct, ok := rg.odb.AccountByName(user)
	if !ok {
		return false
	}
	return rg.odb.EffectiveCaps(acct.Name, repo).HasAny("gjoh2r")
}

// SyncReadGate is the sync protocol's whole visibility answer: the
// xfer.ReadGate installed on the fronter. It takes the SAME answer
// CanRead gives the web pages, refined by a separate question mirroring
// asks that ordinary reading does not — may other people keep a copy
// that follows this repository, not just may THIS caller see it once.
// A private repository answers no regardless of the mirrorable flag
// (MayMirror enforces that); either refusal returns the same "not
// found" a caller cannot distinguish from a repository that simply
// is not there (RFC 0004's invisible-not-forbidden posture).
func (rg *registrar) SyncReadGate(req *http.Request, repo, login string) bool {
	ctx := syncIdentityCtx(req, rg.tokenDoors())
	// The sync protocol's own credential counts as identity here.
	//
	// identityCtx reads HTTP Basic, which a browser and curl send and a
	// sync client does not: fossil and zit authenticate with the login
	// card, inside the body. So a private repository admitted anyone
	// with Basic and refused every real client, including the
	// repository's owner — "private" meaning "unclonable by everybody"
	// (b112b205). xfer has already verified this name against the
	// repository's shared secret; an unverified or absent card arrives
	// as "".
	if login != "" && ops.User(ctx) == "" {
		ctx = ops.WithUser(ctx, login)
		// A card signed with a token reads as the token, ceiling and
		// repositories included (RFC 0021). A token gone since the card
		// verified reads as nothing.
		if id := xfer.LoginToken(req); id != "" && rg.odb != nil {
			tok, ok := rg.odb.LiveToken(login, id)
			if !ok {
				return false
			}
			ctx = rg.tokenDoors().tokenCtx(ctx, login, tok)
		}
	}
	if !rg.CanRead(ctx, repo) {
		// Says WHO was refused, which separates "nobody identified
		// themselves" from "this account may not read this" — the two
		// cases a bare 404 makes identical, and the pair that kept a
		// private repository unclonable while every measurement from
		// outside said the same thing.
		who := ops.User(ctx)
		if who == "" {
			who = "an unidentified caller"
		}
		log.Printf("sync: %s: read refused: %s may not read it", repo, who)
		return false
	}
	if rg.odb == nil {
		return true
	}
	// MayMirror asks "may OTHER PEOPLE keep a copy of this", and it
	// answers no for every private repository by construction. Asked of
	// every caller it locked the door on the way in: a private
	// repository could not be cloned or synced by anyone at all,
	// including the account that owns it, which is not a policy anyone
	// chose — it is "other people" quietly meaning "everyone".
	//
	// A caller who has identified themselves and already satisfied
	// CanRead is not other people. They are someone this repository
	// admits, so the third-party question does not apply to them and
	// the flag is not theirs to be refused by.
	//
	// Anonymous callers still meet it unchanged: that is who the flag
	// was written for, and a public repository with mirroring off still
	// answers no to a stranger.
	if ops.User(ctx) != "" {
		return true
	}
	ok, err := rg.odb.MayMirror(repo)
	if err != nil {
		log.Printf("sync: mirrorable check for %s: %v", repo, err)
		return false
	}
	return ok
}

// 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.
// syncIdentityCtx is identityCtx for the SYNC lane, where the Basic
// password slot may also carry the account's sync token.
//
// A clone's first round has no login card to sign — the client does
// not know the project code yet — so fossil answers the 401 with Basic
// and zit retries with Basic, each putting the credential it holds in
// the password slot. For a client set up with the sync token, that is
// the token; the account password never reaches a sync client on
// purpose (a push signed with it is refused). Read by identityCtx
// alone, the token was a wrong password: an identified caller nobody
// recognised, told 404, and every private repository was unclonable by
// the token minted for it, however correctly the gate would have judged
// the account (2bb5659e, measured by jva).
//
// Only the sync lane takes the token here. The API's Basic lane is
// identityCtx itself and still wants the password: the token is the
// credential for sync, and is not widened into a second password.
//
// RFC 0021: any live sync-scoped token of the account's opens it, by id
// or, for a credential minted before tokens had ids, by its value; the
// token rides the context so the visibility answer is the token's,
// ceiling and repository list included.
func syncIdentityCtx(req *http.Request, doors *tokenDoors) context.Context {
	if doors == nil {
		return req.Context()
	}
	ctx := identityCtx(req, doors.db)
	if ops.User(ctx) != "" || doors.db == nil {
		return ctx
	}
	tok, p := doors.resolve(req, oltp.ScopeSync)
	if p != resolved {
		return ctx
	}
	user, _, _ := req.BasicAuth()
	return doors.tokenCtx(ctx, user, tok)
}

// tokenDoors is rg.doors, or an uncounted resolver over rg.odb.
func (rg *registrar) tokenDoors() *tokenDoors {
	if rg.doors != nil {
		return rg.doors
	}
	return &tokenDoors{db: rg.odb}
}

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)
	}

	// Read before Unregister, which forgets it. A repository that was
	// never served is asked its store directly.
	rg.codeMu.Lock()
	code := rg.codes[name]
	rg.codeMu.Unlock()
	if code == "" {
		code, _ = xfer.ProjectCode(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
	}
	// What the repository left outside the forge database goes before
	// its registration does, so a failure here leaves the name
	// registered and the delete retryable, rather than a new owner
	// inheriting it (quarry#294).
	if err := rg.forget(name, code); err != nil {
		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)
	}
	removeErr := os.RemoveAll(row.Path)
	// Again, now that the caps the mint form is gated on are gone: a
	// token minted between the first sweep and the purge would
	// otherwise be the one the next repository of this name inherits.
	// Whatever became of the store — the name is free either way.
	if err := rg.revokeWorkerTokens(name); err != nil {
		return errors.Join(removeErr, fmt.Errorf("delete %s: the repository is gone, but a worker token "+
			"minted during the delete may still be live — revoke worker group %s's tokens at the engine: %w",
			name, ci.GroupFor(name), err))
	}
	return removeErr
}

// forget erases what a repository being deleted holds outside the forge
// database: the engine's worker tokens scoped to its group, its chat
// channel, its ingest cursors, consumers and store events, and its
// attention events, event log and tallies.
func (rg *registrar) forget(name, code string) error {
	if err := rg.revokeWorkerTokens(name); err != nil {
		return fmt.Errorf("delete %s: worker tokens: %w", name, err)
	}
	if code != "" && !rg.codeServedElsewhere(name, code) {
		if rg.deps != nil && rg.deps.Chat != nil {
			if err := rg.deps.Chat.PurgeChannel(code); err != nil {
				return fmt.Errorf("delete %s: %w", name, err)
			}
		}
		if err := rg.odb.PurgeRobotLines(code); err != nil {
			return fmt.Errorf("delete %s: robot lines: %w", name, err)
		}
	}
	if err := ingest.Forget(rg.ev, name, rg.node); err != nil {
		return fmt.Errorf("delete %s: %w", name, err)
	}
	if err := rg.ev.ForgetRepo(name); err != nil {
		return fmt.Errorf("delete %s: %w", name, err)
	}
	return nil
}

// revokeWorkerTokens revokes every live engine token scoped to the
// repository's worker group. The group is derived from the name, so a
// token left live would claim work in any later repository of the same
// name — and satisfy its required checks. A token scoped to several
// groups goes too: the engine cannot narrow one, and keeping it keeps
// that reach.
//
// A configured engine that cannot list its tokens is a refusal, for any
// reason: quarry cannot tell "none exist" from "none it can see", and
// guessing the first is how a live token outlives its repository. Only
// a forge with no engine at all has nothing to revoke.
func (rg *registrar) revokeWorkerTokens(name string) error {
	if !rg.engine.Enabled() {
		return nil
	}
	tokens, err := rg.engine.WorkerTokens()
	if err != nil {
		return fmt.Errorf("%w: cannot list the CI engine's worker tokens to revoke %s's (%v); "+
			"give quarry the engine's admin token (--dagnats-token), or revoke the tokens scoped "+
			"to worker group %s at the engine, then delete again",
			ops.ErrUnavailable, name, err, ci.GroupFor(name))
	}
	group := ci.GroupFor(name)
	// A site admin's fleet token reaches the group by task type instead:
	// a scope admits the whole type or a namespace it continues after a
	// dot, and the repository's task types are <group>.<run>.
	reaches := func(t ci.WorkerToken) bool {
		if slices.Contains(t.WorkerGroups, group) {
			return true
		}
		return slices.ContainsFunc(t.TaskTypePrefixes, func(p string) bool {
			return p == group || strings.HasPrefix(p, group+".")
		})
	}
	for _, t := range tokens {
		if t.RevokedAt != "" || !reaches(t) {
			continue
		}
		if err := rg.engine.RevokeWorkerToken(t.ID); err != nil {
			return err
		}
	}
	return nil
}

// codeServedElsewhere reports whether a repository other than name is
// registered on the same project code, and so shares its chat channel.
// Every registered repository counts, served or not: one that is not
// served now has no entry in codes, so its store is asked.
func (rg *registrar) codeServedElsewhere(name, code string) bool {
	rg.codeMu.Lock()
	known := make(map[string]string, len(rg.codes))
	for other, c := range rg.codes {
		known[other] = c
	}
	rg.codeMu.Unlock()
	for other, c := range known {
		if other != name && c == code {
			return true
		}
	}
	rows, err := rg.odb.Repos()
	if err != nil {
		// Unsure is shared: leaving a channel costs some messages,
		// emptying another repository's costs its conversation.
		return true
	}
	for _, row := range rows {
		if _, served := known[row.Name]; served || row.Name == name {
			continue
		}
		if c, err := xfer.ProjectCode(row.Path); err == nil && c == code {
			return true
		}
	}
	return false
}

// 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
}

// storeStamp is a store file's identity for the poll loop: the store
// and its write-ahead log, by size and mtime. Reads — zit frames, ls,
// cat — leave both alone; every write moves one of them.
type storeStamp struct {
	ok            bool
	size, walSize int64
	mod, walMod   time.Time
}

// stampStore stamps the store under path (a checkout directory holding
// .zit, or the store file itself). A store the stat cannot see is not
// ok, and the loop polls it every tick as it always did — going quiet
// on a store that is fine but named differently would be the worse
// failure.
func stampStore(path string) storeStamp {
	file := path
	if fi, err := os.Stat(path); err == nil && fi.IsDir() {
		file = filepath.Join(path, ".zit")
	}
	var st storeStamp
	fi, err := os.Stat(file)
	if err != nil {
		return st
	}
	st.ok = true
	st.size, st.mod = fi.Size(), fi.ModTime()
	if wal, err := os.Stat(file + "-wal"); err == nil {
		st.walSize, st.walMod = wal.Size(), wal.ModTime()
	}
	return st
}

// originCheckURL is the public page the promise is checked on, or ""
// when the forge does not know its own public address — in which case
// there is nothing to check FROM outside, and the dep is skipped.
//
// The root, because that is the page most readers actually receive. A
// redirect to a sign-in page is followed and is still a page a reader
// gets, which is the thing being asserted about.
func originCheckURL(origin string) string {
	// Space first, then the slash: the other order leaves a trailing
	// slash attached to a padded value and builds "https://host//".
	origin = strings.TrimSuffix(strings.TrimSpace(origin), "/")
	if origin == "" {
		return ""
	}
	return origin + "/"
}

// claim records the deed and the namespace for a repository that has
// just been registered.
//
// The creator owns what they create, and a repository created by a
// person starts in that person's own namespace -- the personal case
// needs no arranging, which is the whole point of deriving the path
// from the owner. Putting it under a team is a separate, later act
// with its own permission check.
//
// A repository created with no author is forge-held: an operator
// serving a store rather than a person hosting a project. It keeps an
// empty owner rather than being attributed to whoever happened to run
// the command.
//
// Failure here is logged, not fatal. The repository exists and is
// served either way, and a missing deed is recoverable by assigning
// one; refusing the creation because the bookkeeping row would not
// write would be the worse trade.
func (rg *registrar) claim(spec api.RepoSpec) {
	if rg.odb == nil || spec.Author == "" {
		return
	}
	if err := rg.odb.SetOwner(spec.Name, spec.Author); err != nil {
		log.Printf("repo.add %s: recording owner %s: %v", spec.Name, spec.Author, err)
		return
	}
	if err := rg.odb.SetNamespace(spec.Name, spec.Author, spec.Author, time.Now()); err != nil {
		log.Printf("repo.add %s: recording namespace %s: %v", spec.Name, spec.Author, err)
	}
}

// leaseRefusal reports a write the lease refused as ops.ErrUnavailable,
// the answer every op already turns into "try again". The lease's own
// error stays in the chain: the CI poller must tell "another node
// holds this store", which retrying here never fixes, from a
// coordinator that is briefly unreachable.
func leaseRefusal(err error) error {
	if errors.Is(err, lease.ErrNotHolder) || errors.Is(err, lease.ErrUnavailable) {
		return fmt.Errorf("%w: %w", ops.ErrUnavailable, err)
	}
	return err
}