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 {
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
jobs *jobGate
deps *api.Deps
fronter *xfer.Fronter
ingestLane *work.Lane
bulk *work.Lane
engine *ci.Engine
odb *oltp.DB
// openLandings is, per repository, the open landings the fold loop
// has already seen, so the ones a fold ADDS are the ones that just
// opened. 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]bool
// 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)
}
}
rg.gateSweep(name, ix)
rg.carryForward(name, ix)
rg.landingSweep(name, ix)
}
// 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.pairSweep(name, ix, open)
rg.landingsMu.Lock()
if rg.openLandings == nil {
rg.openLandings = map[string]map[string]bool{}
}
seen, seeded := rg.openLandings[name]
if !seeded {
seen = map[string]bool{}
for _, l := range open {
seen[l.ID] = true
}
rg.openLandings[name] = seen
rg.landingsMu.Unlock()
return
}
var fresh []index.LandingSummary
for _, l := range open {
if !seen[l.ID] {
seen[l.ID] = true
fresh = append(fresh, l)
}
}
rg.landingsMu.Unlock()
for _, l := range fresh {
rg.deps.NotifyOpenedLanding(name, l)
}
}
// 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. That is a landing opening, or its branch moving. 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.
func (rg *registrar) pairSweep(name string, ix *index.Index, open []index.LandingSummary) {
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
}
if latest, err := rg.odb.LatestPairFor(name, tip); err != nil || latest != "" {
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.
func (rg *registrar) TriggerCI(name, checkin 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)
}
// 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)
}
// 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 already
// holds every repository (MayAt) 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) SetPassword(name, password string) error {
return a.db.SetPassword(name, password)
}
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, kept apart from SetPassword's
// admin-driven full session sweep (register.go's account.passwd).
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()
}
// 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
}
// 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,
}, 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 — 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 {
if !rg.IsPrivate(repo) {
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")
}
// 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.odb)
// 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)
}
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.
func syncIdentityCtx(req *http.Request, db *oltp.DB) context.Context {
ctx := identityCtx(req, db)
if ops.User(ctx) != "" || db == nil {
return ctx
}
user, pass, ok := req.BasicAuth()
if !ok || user == "" || !db.VerifySyncToken(user, pass) {
return ctx
}
return ops.WithUser(ctx, user)
}
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)
}
if err := os.RemoveAll(row.Path); err != nil {
return err
}
// 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.
if err := rg.revokeWorkerTokens(name); err != nil {
return 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 nil
}
// forget erases what a repository being deleted holds outside the forge
// database: the engine's worker tokens scoped to its group, its chat
// channel, 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 := 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.
func (rg *registrar) revokeWorkerTokens(name string) error {
if !rg.engine.Enabled() {
return nil
}
tokens, err := rg.engine.WorkerTokens()
if ci.TokensUnmanaged(err) {
return nil
}
if err != nil {
return err
}
group := ci.GroupFor(name)
for _, t := range tokens {
if t.RevokedAt != "" || !slices.Contains(t.WorkerGroups, group) {
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.
func (rg *registrar) codeServedElsewhere(name, code string) bool {
rg.codeMu.Lock()
defer rg.codeMu.Unlock()
for other, c := range rg.codes {
if other != name && 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
}